{-# LANGUAGE AllowAmbiguousTypes #-}
{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DerivingVia #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE QuasiQuotes #-}

-- | The job queue operations, over any 'MonadArbiter' backend.
module Arbiter.Core.Operations
  ( -- * Job Insertion
    insertJob
  , insertJobStamped
  , insertJobTreeNodeStamped
  , insertJobTreeLeavesStamped
  , insertJobsBatch
  , insertJobsBatchStamped
  , insertJobsBatch_
  , TraceStamp
  , traceStamp
  , insertResult
  , insertResultsBatch
  , getResultsByParent
  , getDLQChildErrorsByParent
  , persistParentState
  , claimNextVisibleJobs
  , claimNextVisibleJobsAs
  , claimNextVisibleJobsBatched
  , ClaimSql (..)
  , mkClaimSql
  , claimJobsCached
  , claimJobsBatchedCached
  , addRateLimitTokens
  , pruneRateLimitBuckets
  , resetRateLimitBuckets
  , wakeThrottledJobs
  , wakeThrottledJobsForKey
  , listRateLimitPolicies
  , getRateLimitPolicy
  , rateLimitPolicyExists
  , listRateLimitBuckets
  , listConcurrencyPolicies
  , getConcurrencyPolicy
  , listConcurrencyKeys
  , updateRateLimitPolicyOverrides
  , updateConcurrencyPolicyOverrides
  , pruneConcurrencyKeys
  , reconcileConcurrencyCounts
  , reconcileConcurrencyCountsIfStale
  , reconcileAndPruneConcurrency
  , ackJob
  , ackJobInner
  , ackJobsBatch
  , ackJobsBatchInner
  , lockJobParents
  , lockJobTrees
  , lockJobTreesFromRoot
  , TreeLocks (..)
  , archivesOnAck
  , setVisibilityTimeout
  , setVisibilityTimeoutBatch
  , VisibilityUpdateInfo (..)
  , updateJobForRetry
  , nackJob
  , nackJobsBatch
  , moveToDLQ
  , moveToDLQFields
  , moveToDLQBatch
  , retryFromDLQ
  , dlqJobExists
  , listDLQJobs
  , listDLQJobsByParent
  , countDLQJobsByParent
  , deleteDLQJob
  , deleteDLQJobsBatch
  , deleteCancelledJobs

    -- * Completed-Job Archive
  , listArchiveJobs
  , listArchiveFiltered
  , getArchivedJobById
  , listArchivedJobsByGroupKey
  , countArchiveFiltered
  , purgeArchives
  , deleteArchiveJob
  , deleteArchiveJobsBatch
  , reEnqueueFromArchive
  , updateArchiveResult
  , updateArchiveResultsBatch

    -- * Filtered Query Operations
  , Tmpl.JobFilter (..)
  , buildWhereClause
  , listJobsFiltered
  , listJobsFilteredOrdered
  , listJobsWithStatus
  , countJobsFiltered
  , listDLQFiltered
  , listDLQFilteredOrdered
  , countDLQFiltered

    -- * Admin Operations
  , listJobs
  , jobExists
  , getJobById
  , getJobByIdWithStatus
  , getJobByDedupKey
  , getJobsByGroup
  , cancelJob
  , cancelJobsBatch
  , promoteJob
  , QueueStats (..)
  , statsRowCodec
  , queueStatusCounts
  , QueueOverview (..)
  , getQueueStats
  , getAllQueueStats

    -- * Count Operations
  , countJobs
  , countJobsByGroup
  , countDLQJobs

    -- * Parent-Child Operations
  , getJobsByParent
  , countJobsByParent
  , countChildrenBatch
  , countDLQChildrenBatch

    -- * Job Dependency Operations
  , pauseChildren
  , resumeChildren
  , cancelJobCascade
  , cancelJobTree
  , forceCancelJob

    -- * Suspend/Resume Operations
  , suspendJob
  , resumeJob

    -- * Groups Table Operations
  , GroupsCursor (..)
  , GroupsPass (..)
  , refreshGroupsForQueue
  , refreshAllGroups
  , refreshAllGroupsFully
  , sweepExhaustedJobs
  , sweepCancelledJobs

    -- * Cron Schedule Operations
  , upsertCronDefault
  , listCronSchedules
  , getCronScheduleByName
  , updateCronSchedule
  , touchCronLastFired
  , touchCronChecked
  , tryFireCronGate
  , tryAcquireCronLeader
  , RunRequestOutcome (..)
  , requestCronRun
  , claimCronRun
  , touchCronManualRun
  , pendingCronRuns

    -- * Worker Registry Operations
  , registerWorker
  , heartbeatWorker
  , setWorkerPaused
  , markWorkerShuttingDown
  , deregisterWorker
  , workerRegistered
  , listWorkers
  , sweepStaleWorkers

    -- * Queue Operations
  , ensureQueue
  , setQueuePaused
  , getQueue
  , listQueues

    -- * Global Gate Operations
  , runGated
  , runGatedBounded
  , runGatedShared
  , runGatedState
  , runGatedStateBounded
  , setLocalStatementTimeout
  , micros
  , gateNameFor
  , Shared (..)

    -- * Internal Operations
  , getParentStateSnapshot
  , readChildResultsRaw
  , mergeRawChildResults
  ) where

import Control.Monad (foldM, join, unless, void, when)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Data.Aeson
  ( FromJSON (..)
  , Result (..)
  , ToJSON (..)
  , Value
  , fromJSON
  , object
  , withObject
  , (.!=)
  , (.:)
  , (.:?)
  , (.=)
  )
import Data.Bifunctor (bimap, first)
import Data.Either (fromRight, isLeft, partitionEithers)
import Data.Foldable (for_, toList, traverse_)
import Data.Int (Int64)
import Data.IntMap qualified as IntMap
import Data.List (groupBy, sortOn)
import Data.List.NonEmpty (NonEmpty (..))
import Data.List.NonEmpty qualified as NE
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Maybe (catMaybes, fromMaybe, listToMaybe, mapMaybe)
import Data.Monoid (Ap (..), Sum (..))
import Data.Proxy (Proxy (..))
import Data.Sequence ((|>))
import Data.Sequence qualified as Seq
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (NominalDiffTime, UTCTime)
import Data.UUID.Types (UUID)
import Data.UUID.Types qualified as UUID
import GHC.Generics (Generic)
import UnliftIO (MonadUnliftIO, tryAny)

import Arbiter.Core.Codec
  ( Col (..)
  , JobWriteSource (..)
  , RowCodec
  , col
  , jobCodec
  , jobRowCodec
  , ncol
  , pval
  )
import Arbiter.Core.Concurrency.Spec
  ( ConcurrencyKey (..)
  , HasConcurrency
  , concurrencyFor
  , concurrencyKeyText
  , runConcurrencyFor
  )
import Arbiter.Core.Concurrency.Stats (ConcurrencyKeyView, ConcurrencyPolicyUpdate (..), ConcurrencyPolicyView)
import Arbiter.Core.CronSchedule (CronScheduleRow, CronScheduleUpdate (..))
import Arbiter.Core.Exceptions (throwParsing)
import Arbiter.Core.Job.Archive qualified as Archive
import Arbiter.Core.Job.DLQ qualified as DLQ
import Arbiter.Core.Job.Kind (HasKind, kindOf)
import Arbiter.Core.Job.Schema (SchemaName, TableName)
import Arbiter.Core.Job.Types
  ( ClaimSeq
  , DedupKey (ReplaceDuplicate)
  , JobId
  , JobPayload
  , JobRead
  , JobStatus (..)
  , JobWrite
  , PayloadColumns (..)
  , archiveFor
  , attempts
  , claimSeq
  , claimedBy
  , dedupParts
  , groupKey
  , isRollup
  , jobStatusFromText
  , jobStatusToText
  , mapPayload
  , parentId
  , payload
  , primaryKey
  )
import Arbiter.Core.Job.Types qualified as JT
import Arbiter.Core.MonadArbiter (MonadArbiter, withDbTransaction)
import Arbiter.Core.MonadArbiter qualified as MA
import Arbiter.Core.Operations.Gates
  ( Shared (..)
  , gateNameFor
  , micros
  , runGated
  , runGatedBounded
  , runGatedShared
  , runGatedState
  , runGatedStateBounded
  , setLocalStatementTimeout
  )
import Arbiter.Core.Operations.Workers
  ( deregisterWorker
  , heartbeatWorker
  , listWorkers
  , markWorkerShuttingDown
  , registerWorker
  , setWorkerPaused
  , sweepStaleWorkers
  , workerRegistered
  )
import Arbiter.Core.Queues (QueueRow)
import Arbiter.Core.RateLimit.Spec
  ( HasRateLimit
  , RateLimitKey (..)
  , rateLimitCost
  , rateLimitFor
  , rateLimitKeyText
  , runRateLimitFor
  )
import Arbiter.Core.RateLimit.Stats (RateLimitBucketView, RateLimitPolicyUpdate (..), RateLimitPolicyView)
import Arbiter.Core.Selector (usesAnyPolicy)
import Arbiter.Core.Sql.Archive qualified as Tmpl
import Arbiter.Core.Sql.Claim qualified as Claim
import Arbiter.Core.Sql.Concurrency qualified as Tmpl
import Arbiter.Core.Sql.Cron qualified as Tmpl
import Arbiter.Core.Sql.DLQ qualified as Tmpl
import Arbiter.Core.Sql.Groups qualified as Tmpl
import Arbiter.Core.Sql.Insert (batchFrag, insertFrag)
import Arbiter.Core.Sql.Jobs qualified as Tmpl
import Arbiter.Core.Sql.Lifecycle qualified as Tmpl
import Arbiter.Core.Sql.QQ qualified as QQ
import Arbiter.Core.Sql.Query qualified as Q
import Arbiter.Core.Sql.Queues qualified as Tmpl
import Arbiter.Core.Sql.RateLimit qualified as Tmpl
import Arbiter.Core.Sql.Stats qualified as Tmpl
import Arbiter.Core.Sql.Tree qualified as Tmpl
import Arbiter.Core.Trace (currentTraceContext, stampTraceContext)

decodePayload :: (JobPayload payload, MonadArbiter m) => JobRead Value -> m (JobRead payload)
decodePayload :: forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
JobRead Value -> m (JobRead payload)
decodePayload JobRead Value
job = case Value -> Result payload
forall a. FromJSON a => Value -> Result a
fromJSON (JobRead Value -> Value
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> payload
payload JobRead Value
job) of
  Success payload
decoded -> JobRead payload -> m (JobRead payload)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (JobRead payload -> m (JobRead payload))
-> JobRead payload -> m (JobRead payload)
forall a b. (a -> b) -> a -> b
$ (Value -> payload) -> JobRead Value -> JobRead payload
forall payload payload' key q insertedAt adm.
(payload -> payload')
-> Job payload key q insertedAt adm
-> Job payload' key q insertedAt adm
mapPayload (payload -> Value -> payload
forall a b. a -> b -> a
const payload
decoded) JobRead Value
job
  Error String
err -> Text -> m (JobRead payload)
forall (m :: * -> *) a. MonadIO m => Text -> m a
throwParsing (Text -> m (JobRead payload)) -> Text -> m (JobRead payload)
forall a b. (a -> b) -> a -> b
$ Text
"Failed to decode job payload: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack String
err

visibilityUpdateCodec :: RowCodec VisibilityUpdateInfo
visibilityUpdateCodec :: RowCodec VisibilityUpdateInfo
visibilityUpdateCodec =
  Int64
-> Bool
-> Maybe Int64
-> Bool
-> Bool
-> Maybe UUID
-> VisibilityUpdateInfo
VisibilityUpdateInfo
    (Int64
 -> Bool
 -> Maybe Int64
 -> Bool
 -> Bool
 -> Maybe UUID
 -> VisibilityUpdateInfo)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Bool
      -> Maybe Int64
      -> Bool
      -> Bool
      -> Maybe UUID
      -> VisibilityUpdateInfo)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"id" Col Int64
CInt8
    Ap
  NullCol
  (Bool
   -> Maybe Int64
   -> Bool
   -> Bool
   -> Maybe UUID
   -> VisibilityUpdateInfo)
-> Ap NullCol Bool
-> Ap
     NullCol
     (Maybe Int64 -> Bool -> Bool -> Maybe UUID -> VisibilityUpdateInfo)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Bool -> Ap NullCol Bool
forall a. Text -> Col a -> RowCodec a
col Text
"was_heartbeated" Col Bool
CBool
    Ap
  NullCol
  (Maybe Int64 -> Bool -> Bool -> Maybe UUID -> VisibilityUpdateInfo)
-> Ap NullCol (Maybe Int64)
-> Ap NullCol (Bool -> Bool -> Maybe UUID -> VisibilityUpdateInfo)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol (Maybe Int64)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"current_db_claim_seq" Col Int64
CInt8
    Ap NullCol (Bool -> Bool -> Maybe UUID -> VisibilityUpdateInfo)
-> Ap NullCol Bool
-> Ap NullCol (Bool -> Maybe UUID -> VisibilityUpdateInfo)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Bool -> Ap NullCol Bool
forall a. Text -> Col a -> RowCodec a
col Text
"cancel_requested" Col Bool
CBool
    Ap NullCol (Bool -> Maybe UUID -> VisibilityUpdateInfo)
-> Ap NullCol Bool
-> Ap NullCol (Maybe UUID -> VisibilityUpdateInfo)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Bool -> Ap NullCol Bool
forall a. Text -> Col a -> RowCodec a
col Text
"suspended" Col Bool
CBool
    Ap NullCol (Maybe UUID -> VisibilityUpdateInfo)
-> Ap NullCol (Maybe UUID) -> RowCodec VisibilityUpdateInfo
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col UUID -> Ap NullCol (Maybe UUID)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"claimed_by" Col UUID
CUuid

parentCountCodec :: RowCodec (Int64, (Int64, Int64))
parentCountCodec :: RowCodec (Int64, (Int64, Int64))
parentCountCodec =
  (\Int64
pid Int64
cnt Int64
paused -> (Int64
pid, (Int64
cnt, Int64
paused)))
    (Int64 -> Int64 -> Int64 -> (Int64, (Int64, Int64)))
-> Ap NullCol Int64
-> Ap NullCol (Int64 -> Int64 -> (Int64, (Int64, Int64)))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"parent_id" Col Int64
CInt8
    Ap NullCol (Int64 -> Int64 -> (Int64, (Int64, Int64)))
-> Ap NullCol Int64
-> Ap NullCol (Int64 -> (Int64, (Int64, Int64)))
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"count" Col Int64
CInt8
    Ap NullCol (Int64 -> (Int64, (Int64, Int64)))
-> Ap NullCol Int64 -> RowCodec (Int64, (Int64, Int64))
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"count_suspended" Col Int64
CInt8

-- | The @WHERE@ clause a set of filters narrows a listing by. Empty for no filters.
buildWhereClause :: [Tmpl.JobFilter] -> Q.Query ()
buildWhereClause :: [JobFilter] -> Query ()
buildWhereClause [] = Query ()
forall a. Monoid a => a
mempty
buildWhereClause [JobFilter]
filters = Text -> Query ()
Q.raw Text
"WHERE " Query () -> Query () -> Query ()
forall a. Semigroup a => a -> a -> a
<> Text -> [Query ()] -> Query ()
Q.sepBy Text
" AND " ((JobFilter -> Query ()) -> [JobFilter] -> [Query ()]
forall a b. (a -> b) -> [a] -> [b]
map JobFilter -> Query ()
filterToClause [JobFilter]
filters)

filterToClause :: Tmpl.JobFilter -> Q.Query ()
filterToClause :: JobFilter -> Query ()
filterToClause (Tmpl.FilterGroupKey Text
key) = [QQ.sql|group_key = #{key :: CText}|]
filterToClause (Tmpl.FilterParentId Int64
pid) = [QQ.sql|parent_id = #{pid :: CInt8}|]
filterToClause JobFilter
Tmpl.FilterRootsOnly = Text -> Query ()
Q.raw Text
"parent_id IS NULL"
filterToClause (Tmpl.FilterStatus JobStatus
status) = [QQ.sql|status = #{statusText :: CText}|]
  where
    statusText :: Text
statusText = JobStatus -> Text
jobStatusToText JobStatus
status
filterToClause (Tmpl.FilterId Int64
jobId) = [QQ.sql|id = #{jobId :: CInt8}|]
filterToClause (Tmpl.FilterJobId Int64
jobId) = [QQ.sql|job_id = #{jobId :: CInt8}|]
filterToClause (Tmpl.FilterClaimedBy UUID
workerId) = [QQ.sql|claimed_by = #{workerId :: CUuid}|]
filterToClause (Tmpl.FilterKind Text
kind) = [QQ.sql|kind = #{kind :: CText}|]
filterToClause (Tmpl.FilterRateLimitPrefix Text
prefix) = [QQ.sql|rate_limit_prefix = #{prefix :: CText}|]
filterToClause (Tmpl.FilterConcurrencyPrefix Text
prefix) = [QQ.sql|concurrency_prefix = #{prefix :: CText}|]
filterToClause (Tmpl.FilterInsertedAfter UTCTime
time) = [QQ.sql|inserted_at >= #{time :: CTimestamptz}|]
filterToClause (Tmpl.FilterInsertedBefore UTCTime
time) = [QQ.sql|inserted_at < #{time :: CTimestamptz}|]
filterToClause (Tmpl.FilterCompletedAfter UTCTime
time) = [QQ.sql|completed_at >= #{time :: CTimestamptz}|]
filterToClause (Tmpl.FilterCompletedBefore UTCTime
time) = [QQ.sql|completed_at < #{time :: CTimestamptz}|]
filterToClause (Tmpl.FilterPayloadText Text
needle) = [QQ.sql|payload::text ILIKE #{pat :: CText} ESCAPE '\' |]
  where
    pat :: Text
pat = Text
"%" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
likeEscape Text
needle Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"%"

-- | Escape the LIKE metacharacters in a substring searched for literally.
likeEscape :: Text -> Text
likeEscape :: Text -> Text
likeEscape = (Char -> Text) -> Text -> Text
T.concatMap Char -> Text
escape
  where
    escape :: Char -> Text
escape Char
ch = if Char
ch Char -> String -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` String
likeMeta then String -> Text
T.pack [Char
'\\', Char
ch] else Char -> Text
T.singleton Char
ch

-- | The characters LIKE reads as pattern syntax, plus its own escape character.
likeMeta :: String
likeMeta :: String
likeMeta = String
"\\%_"

-- | Run a single-row count @Query@, throwing a parse error on unexpected results.
countStrict :: (MonadArbiter m) => Text -> Q.Query Int64 -> m Int64
countStrict :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict Text
label Query Int64
query = do
  rows <- Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery Query Int64
query
  case rows of
    [Int64
count] -> Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
count
    [Int64]
_ -> Text -> m Int64
forall (m :: * -> *) a. MonadIO m => Text -> m a
throwParsing (Text -> m Int64) -> Text -> m Int64
forall a b. (a -> b) -> a -> b
$ Text
label Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
": unexpected result"

-- | Run a single-row count @Query@, returning 0 on an empty or unexpected result.
countOr0 :: (MonadArbiter m) => Q.Query Int64 -> m Int64
countOr0 :: forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 = ([Int64] -> Int64) -> m [Int64] -> m Int64
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap [Int64] -> Int64
singleCount (m [Int64] -> m Int64)
-> (Query Int64 -> m [Int64]) -> Query Int64 -> m Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery

countOr0Prepared :: (MonadArbiter m) => Q.Query Int64 -> m Int64
countOr0Prepared :: forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0Prepared = ([Int64] -> Int64) -> m [Int64] -> m Int64
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap [Int64] -> Int64
singleCount (m [Int64] -> m Int64)
-> (Query Int64 -> m [Int64]) -> Query Int64 -> m Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQueryPrepared

singleCount :: [Int64] -> Int64
singleCount :: [Int64] -> Int64
singleCount [Int64
count] = Int64
count
singleCount [Int64]
_ = Int64
0

-- | Take a transaction-scoped advisory lock keyed by a @schema.table@ string and a job id.
advisoryXactLockSQL :: Text -> Int64 -> Q.Query (Maybe Text)
advisoryXactLockSQL :: Text -> Int64 -> Query (Maybe Text)
advisoryXactLockSQL Text
key Int64
pid =
  [QQ.sql|SELECT pg_advisory_xact_lock(hashtextextended(#{key :: CText}, #{pid :: CInt8}))::text AS @{result :: Maybe CText}|]

-- | 'advisoryXactLockSQL' over many ids, ascending, in one round trip.
advisoryXactLockManySQL :: Text -> [Int64] -> Q.Query (Maybe Text)
advisoryXactLockManySQL :: Text -> [Int64] -> Query (Maybe Text)
advisoryXactLockManySQL Text
key [Int64]
pids =
  [QQ.sql|SELECT pg_advisory_xact_lock(hashtextextended(#{key :: CText}, id))::text AS @{result :: Maybe CText} FROM unnest(#{pids :: [CInt8]}::bigint[]) AS job_ids(id) ORDER BY id|]

-- | The columns a job derives from its payload.
payloadColumns
  :: forall payload
   . (HasConcurrency payload, HasKind payload, HasRateLimit payload)
  => payload
  -> PayloadColumns
payloadColumns :: forall payload.
(HasConcurrency payload, HasKind payload, HasRateLimit payload) =>
payload -> PayloadColumns
payloadColumns payload
payloadValue =
  let rlKey :: Maybe RateLimitKey
rlKey = payload -> RateLimitFor payload -> Maybe RateLimitKey
forall payload.
payload -> RateLimitFor payload -> Maybe RateLimitKey
runRateLimitFor payload
payloadValue (forall payload. HasRateLimit payload => RateLimitFor payload
rateLimitFor @payload)
      ccKey :: Maybe ConcurrencyKey
ccKey = payload -> ConcurrencyFor payload -> Maybe ConcurrencyKey
forall payload.
payload -> ConcurrencyFor payload -> Maybe ConcurrencyKey
runConcurrencyFor payload
payloadValue (forall payload. HasConcurrency payload => ConcurrencyFor payload
concurrencyFor @payload)
   in PayloadColumns
        { pcKind :: Maybe Text
pcKind = payload -> Maybe Text
forall payload. HasKind payload => payload -> Maybe Text
kindOf payload
payloadValue
        , pcRateLimitKey :: Maybe Text
pcRateLimitKey = RateLimitKey -> Text
rateLimitKeyText (RateLimitKey -> Text) -> Maybe RateLimitKey -> Maybe Text
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe RateLimitKey
rlKey
        , pcRateLimitPrefix :: Maybe Text
pcRateLimitPrefix = RateLimitKey -> Text
rlkPrefix (RateLimitKey -> Text) -> Maybe RateLimitKey -> Maybe Text
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe RateLimitKey
rlKey
        , pcRateLimitCost :: Double
pcRateLimitCost = payload -> Double
forall payload. HasRateLimit payload => payload -> Double
rateLimitCost payload
payloadValue
        , pcConcurrencyKey :: Maybe Text
pcConcurrencyKey = ConcurrencyKey -> Text
concurrencyKeyText (ConcurrencyKey -> Text) -> Maybe ConcurrencyKey -> Maybe Text
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe ConcurrencyKey
ccKey
        , pcConcurrencyPrefix :: Maybe Text
pcConcurrencyPrefix = ConcurrencyKey -> Text
ckPrefix (ConcurrencyKey -> Text) -> Maybe ConcurrencyKey -> Maybe Text
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe ConcurrencyKey
ccKey
        }

-- | What an insert path puts on its jobs, carrying the ambient trace context.
type TraceStamp payload = JobWrite payload -> JobWrite payload

-- | The stamp for the context in scope. One read covers every job inserted under it.
traceStamp :: (MonadIO m) => m (TraceStamp payload)
traceStamp :: forall (m :: * -> *) payload. MonadIO m => m (TraceStamp payload)
traceStamp = Maybe TraceContext -> JobWrite payload -> JobWrite payload
forall payload.
Maybe TraceContext -> JobWrite payload -> JobWrite payload
stampTraceContext (Maybe TraceContext -> JobWrite payload -> JobWrite payload)
-> m (Maybe TraceContext)
-> m (JobWrite payload -> JobWrite payload)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO (Maybe TraceContext) -> m (Maybe TraceContext)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO (Maybe TraceContext)
currentTraceContext

stampedRow
  :: (JobPayload payload)
  => TraceStamp payload
  -> JobWrite payload
  -> JobWriteSource payload
stampedRow :: forall payload.
JobPayload payload =>
TraceStamp payload -> JobWrite payload -> JobWriteSource payload
stampedRow TraceStamp payload
stamp JobWrite payload
job = TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> JobWriteSource payload
forall payload.
JobPayload payload =>
TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> JobWriteSource payload
internalStampedRow TraceStamp payload
stamp Maybe Int64
forall a. Maybe a
Nothing Maybe Value
forall a. Maybe a
Nothing Bool
False JobWrite payload
job

internalStampedRow
  :: (JobPayload payload)
  => TraceStamp payload
  -> Maybe Int64
  -> Maybe Value
  -> Bool
  -> JobWrite payload
  -> JobWriteSource payload
internalStampedRow :: forall payload.
JobPayload payload =>
TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> JobWriteSource payload
internalStampedRow TraceStamp payload
stamp Maybe Int64
parent Maybe Value
state Bool
suspended JobWrite payload
job =
  let stamped :: JobWrite payload
stamped = TraceStamp payload
stamp JobWrite payload
job
      encoded :: Value
encoded = payload -> Value
forall a. ToJSON a => a -> Value
toJSON (JobWrite payload -> payload
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> payload
JT.payload JobWrite payload
stamped)
   in JobWriteSource
        { sourceJob :: JobWrite payload
sourceJob = JobWrite payload
stamped
        , sourceEncoded :: Value
sourceEncoded = Value
encoded
        , sourceColumns :: PayloadColumns
sourceColumns = payload -> PayloadColumns
forall payload.
(HasConcurrency payload, HasKind payload, HasRateLimit payload) =>
payload -> PayloadColumns
payloadColumns (JobWrite payload -> payload
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> payload
JT.payload JobWrite payload
stamped)
        , sourceParentId :: Maybe Int64
sourceParentId = Maybe Int64
parent
        , sourceParentState :: Maybe Value
sourceParentState = Maybe Value
state
        , sourceSuspended :: Bool
sourceSuspended = Bool
suspended
        }

-- | 'insertJob' over a stamp the caller shares across its inserts.
insertJobStamped
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> TraceStamp payload
  -> JobWrite payload
  -> m (Maybe (JobRead payload))
insertJobStamped :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> TraceStamp payload
-> JobWrite payload
-> m (Maybe (JobRead payload))
insertJobStamped Text
schemaName Text
tableName TraceStamp payload
stamp JobWrite payload
job =
  Text
-> Text
-> JobWrite payload
-> JobWriteSource payload
-> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> JobWrite payload
-> JobWriteSource payload
-> m (Maybe (JobRead payload))
insertJobSource Text
schemaName Text
tableName JobWrite payload
job (TraceStamp payload -> JobWrite payload -> JobWriteSource payload
forall payload.
JobPayload payload =>
TraceStamp payload -> JobWrite payload -> JobWriteSource payload
stampedRow TraceStamp payload
stamp JobWrite payload
job)

-- | Internal tree insertion path for engine-owned parent and suspension state.
insertJobTreeNodeStamped
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> TraceStamp payload
  -> Maybe Int64
  -> Maybe Value
  -> Bool
  -> JobWrite payload
  -> m (Maybe (JobRead payload))
insertJobTreeNodeStamped :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> m (Maybe (JobRead payload))
insertJobTreeNodeStamped Text
schemaName Text
tableName TraceStamp payload
stamp Maybe Int64
parent Maybe Value
state Bool
suspended JobWrite payload
job =
  Text
-> Text
-> JobWrite payload
-> JobWriteSource payload
-> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> JobWrite payload
-> JobWriteSource payload
-> m (Maybe (JobRead payload))
insertJobSource Text
schemaName Text
tableName JobWrite payload
job (TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> JobWriteSource payload
forall payload.
JobPayload payload =>
TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> JobWriteSource payload
internalStampedRow TraceStamp payload
stamp Maybe Int64
parent Maybe Value
state Bool
suspended JobWrite payload
job)

insertJobSource
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> JobWrite payload
  -> JobWriteSource payload
  -> m (Maybe (JobRead payload))
insertJobSource :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> JobWrite payload
-> JobWriteSource payload
-> m (Maybe (JobRead payload))
insertJobSource Text
schemaName Text
tableName JobWrite payload
job JobWriteSource payload
source = do
  let valuesFrag :: Query ()
valuesFrag = Codec (JobWriteSource payload) (JobRead Value)
-> JobWriteSource payload -> Query ()
forall s a. Codec s a -> s -> Query ()
insertFrag (Text -> Codec (JobWriteSource payload) (JobRead Value)
forall payload.
Text -> Codec (JobWriteSource payload) (JobRead Value)
jobCodec Text
tableName) JobWriteSource payload
source
      query :: Query (JobRead Value)
query = case JobWrite payload -> Maybe DedupKey
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey JobWrite payload
job of
        Just (ReplaceDuplicate Text
_) -> Text -> Text -> Query () -> Query (JobRead Value)
Tmpl.insertJobReplaceSQL Text
schemaName Text
tableName Query ()
valuesFrag
        Maybe DedupKey
_ -> Text -> Text -> Query () -> Query (JobRead Value)
Tmpl.insertJobSQL Text
schemaName Text
tableName Query ()
valuesFrag

  m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ do
    rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery Query (JobRead Value)
query
    case rawJobs of
      [] -> m (Maybe (JobRead payload))
-> (DedupKey -> m (Maybe (JobRead payload)))
-> Maybe DedupKey
-> m (Maybe (JobRead payload))
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (Text -> m (Maybe (JobRead payload))
forall (m :: * -> *) a. MonadIO m => Text -> m a
throwParsing Text
"insertJob: No rows returned from INSERT") (m (Maybe (JobRead payload))
-> DedupKey -> m (Maybe (JobRead payload))
forall a b. a -> b -> a
const (Maybe (JobRead payload) -> m (Maybe (JobRead payload))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (JobRead payload)
forall a. Maybe a
Nothing)) (JobWrite payload -> Maybe DedupKey
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey JobWrite payload
job)
      (JobRead Value
raw : [JobRead Value]
_) -> JobRead payload -> Maybe (JobRead payload)
forall a. a -> Maybe a
Just (JobRead payload -> Maybe (JobRead payload))
-> m (JobRead payload) -> m (Maybe (JobRead payload))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> JobRead Value -> m (JobRead payload)
forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
JobRead Value -> m (JobRead payload)
decodePayload JobRead Value
raw

-- | Batch-insert direct tree leaves under one parent.
insertJobTreeLeavesStamped
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> TraceStamp payload
  -> Int64
  -> [JobWrite payload]
  -> m [JobRead payload]
insertJobTreeLeavesStamped :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> TraceStamp payload
-> Int64
-> [JobWrite payload]
-> m [JobRead payload]
insertJobTreeLeavesStamped Text
_ Text
_ TraceStamp payload
_ Int64
_ [] = [JobRead payload] -> m [JobRead payload]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
insertJobTreeLeavesStamped Text
schemaName Text
tableName TraceStamp payload
stamp Int64
parent [JobWrite payload]
jobs = do
  let rows :: [JobWriteSource payload]
rows =
        [ TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> JobWriteSource payload
forall payload.
JobPayload payload =>
TraceStamp payload
-> Maybe Int64
-> Maybe Value
-> Bool
-> JobWrite payload
-> JobWriteSource payload
internalStampedRow TraceStamp payload
stamp (Int64 -> Maybe Int64
forall a. a -> Maybe a
Just Int64
parent) Maybe Value
forall a. Maybe a
Nothing Bool
False JobWrite payload
job
        | JobWrite payload
job <- [JobWrite payload] -> [JobWrite payload]
forall payload. [JobWrite payload] -> [JobWrite payload]
dedupBatch [JobWrite payload]
jobs
        ]
      batchSrc :: Query ()
batchSrc = Codec (JobWriteSource payload) (JobRead Value)
-> [JobWriteSource payload] -> Query ()
forall s a. Codec s a -> [s] -> Query ()
batchFrag (Text -> Codec (JobWriteSource payload) (JobRead Value)
forall payload.
Text -> Codec (JobWriteSource payload) (JobRead Value)
jobCodec Text
tableName) [JobWriteSource payload]
rows
  m [JobRead payload] -> m [JobRead payload]
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m [JobRead payload] -> m [JobRead payload])
-> m [JobRead payload] -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ do
    rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query () -> Query (JobRead Value)
Tmpl.insertJobsBatchSQL Text
schemaName Text
tableName Query ()
batchSrc)
    traverse decodePayload rawJobs

-- | Add tokens to a key's bucket, capped at its max. For operator top-ups and
-- manually-refilled policies.
addRateLimitTokens :: (MonadArbiter m) => SchemaName -> RateLimitKey -> Double -> m ()
addRateLimitTokens :: forall (m :: * -> *).
MonadArbiter m =>
Text -> RateLimitKey -> Double -> m ()
addRateLimitTokens Text
schemaName RateLimitKey
key Double
amount =
  m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m Int64 -> m ()) -> m Int64 -> m ()
forall a b. (a -> b) -> a -> b
$
    Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
      (Text -> Text -> Text -> Double -> Query ()
Tmpl.addRateLimitTokensSQL Text
schemaName (RateLimitKey -> Text
rateLimitKeyText RateLimitKey
key) (RateLimitKey -> Text
rlkPrefix RateLimitKey
key) Double
amount)

-- | Delete full, idle rate-limit buckets. Returns the number pruned.
pruneRateLimitBuckets :: (MonadArbiter m) => SchemaName -> NominalDiffTime -> m Int64
pruneRateLimitBuckets :: forall (m :: * -> *).
MonadArbiter m =>
Text -> NominalDiffTime -> m Int64
pruneRateLimitBuckets Text
schemaName NominalDiffTime
idle =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Double -> Query ()
Tmpl.pruneRateLimitBucketsSQL Text
schemaName (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac NominalDiffTime
idle))

-- | Refill every bucket under a prefix to full. Returns the number refilled. Used to
-- build a fixed window from a manual policy plus a cron.
resetRateLimitBuckets :: (MonadArbiter m) => SchemaName -> Text -> m Int64
resetRateLimitBuckets :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
resetRateLimitBuckets Text
schemaName Text
prefix =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Query ()
Tmpl.resetRateLimitBucketsSQL Text
schemaName Text
prefix)

-- | Make a prefix's throttled jobs claimable again across the given queue tables,
-- in one statement. Returns the number woken.
wakeThrottledJobs :: (MonadArbiter m) => SchemaName -> [TableName] -> Text -> m Int64
wakeThrottledJobs :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> Text -> m Int64
wakeThrottledJobs Text
_ [] Text
_ = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
wakeThrottledJobs Text
schemaName [Text]
tableNames Text
prefix =
  Text -> Query Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict Text
"wakeThrottledJobs" (Text -> [Text] -> Text -> Query Int64
Tmpl.wakeThrottledJobsSQL Text
schemaName [Text]
tableNames Text
prefix)

-- | Wake one key's throttled jobs across the given tables, in one statement.
-- Returns the count.
wakeThrottledJobsForKey :: (MonadArbiter m) => SchemaName -> [TableName] -> RateLimitKey -> m Int64
wakeThrottledJobsForKey :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> RateLimitKey -> m Int64
wakeThrottledJobsForKey Text
_ [] RateLimitKey
_ = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
wakeThrottledJobsForKey Text
schemaName [Text]
tableNames RateLimitKey
key =
  Text -> Query Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict
    Text
"wakeThrottledJobsForKey"
    (Text -> [Text] -> Text -> Text -> Query Int64
Tmpl.wakeThrottledJobsForKeySQL Text
schemaName [Text]
tableNames (RateLimitKey -> Text
rlkPrefix RateLimitKey
key) (RateLimitKey -> Text
rateLimitKeyText RateLimitKey
key))

-- | List every policy with its default/override params, bucket aggregates, and
-- live throttled count across the given queue tables.
listRateLimitPolicies :: (MonadArbiter m) => SchemaName -> [TableName] -> m [RateLimitPolicyView]
listRateLimitPolicies :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m [RateLimitPolicyView]
listRateLimitPolicies Text
schemaName [Text]
tableNames =
  Query RateLimitPolicyView -> m [RateLimitPolicyView]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> [Text] -> Query RateLimitPolicyView
Tmpl.listRateLimitPoliciesSQL Text
schemaName [Text]
tableNames)

-- | One prefix's policy view with bucket aggregates and live throttled count.
getRateLimitPolicy :: (MonadArbiter m) => SchemaName -> [TableName] -> Text -> m (Maybe RateLimitPolicyView)
getRateLimitPolicy :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> Text -> m (Maybe RateLimitPolicyView)
getRateLimitPolicy Text
schemaName [Text]
tableNames Text
prefix =
  [RateLimitPolicyView] -> Maybe RateLimitPolicyView
forall a. [a] -> Maybe a
listToMaybe
    ([RateLimitPolicyView] -> Maybe RateLimitPolicyView)
-> m [RateLimitPolicyView] -> m (Maybe RateLimitPolicyView)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query RateLimitPolicyView -> m [RateLimitPolicyView]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> [Text] -> Text -> Query RateLimitPolicyView
Tmpl.getRateLimitPolicySQL Text
schemaName [Text]
tableNames Text
prefix)

-- | Whether a rate-limit policy exists for a prefix.
rateLimitPolicyExists :: (MonadArbiter m) => SchemaName -> Text -> m Bool
rateLimitPolicyExists :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Bool
rateLimitPolicyExists Text
schemaName Text
prefix =
  [Bool] -> Bool
forall (t :: * -> *). Foldable t => t Bool -> Bool
or ([Bool] -> Bool) -> m [Bool] -> m Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query Bool -> m [Bool]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query Bool
Tmpl.rateLimitPolicyExistsSQL Text
schemaName Text
prefix)

-- | List a prefix's buckets with effective max and fill fraction, paginated.
listRateLimitBuckets :: (MonadArbiter m) => SchemaName -> Text -> Int -> Int -> m [RateLimitBucketView]
listRateLimitBuckets :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [RateLimitBucketView]
listRateLimitBuckets Text
schemaName Text
prefix Int
limit Int
offset =
  Query RateLimitBucketView -> m [RateLimitBucketView]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
    (Text -> Text -> Int64 -> Int64 -> Query RateLimitBucketView
Tmpl.listRateLimitBucketsSQL Text
schemaName Text
prefix (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
limit) (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
offset))

-- | Set or clear a policy's override params. Returns rows affected (0 if absent).
updateRateLimitPolicyOverrides :: (MonadArbiter m) => SchemaName -> Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides Text
schemaName Text
prefix (RateLimitPolicyUpdate Maybe (Maybe Double)
mMax Maybe (Maybe Double)
mRefill Maybe (Maybe Double)
mInterval) =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text
-> Maybe (Maybe Double)
-> Maybe (Maybe Double)
-> Maybe (Maybe Double)
-> Text
-> Query ()
Tmpl.updateRateLimitOverridesSQL Text
schemaName Maybe (Maybe Double)
mMax Maybe (Maybe Double)
mRefill Maybe (Maybe Double)
mInterval Text
prefix)

-- | Apply a pool's override-limit patch (retunes every key under the prefix).
-- Returns rows affected.
updateConcurrencyPolicyOverrides :: (MonadArbiter m) => SchemaName -> Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides Text
schemaName Text
prefix (ConcurrencyPolicyUpdate Maybe (Maybe Int32)
mLim) =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Maybe (Maybe Int32) -> Text -> Query ()
Tmpl.updateConcurrencyPolicyOverrideSQL Text
schemaName Maybe (Maybe Int32)
mLim Text
prefix)

-- | List every concurrency pool with its default/override limit and live key and
-- in-flight aggregates.
listConcurrencyPolicies :: (MonadArbiter m) => SchemaName -> m [ConcurrencyPolicyView]
listConcurrencyPolicies :: forall (m :: * -> *).
MonadArbiter m =>
Text -> m [ConcurrencyPolicyView]
listConcurrencyPolicies Text
schemaName =
  Query ConcurrencyPolicyView -> m [ConcurrencyPolicyView]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Query ConcurrencyPolicyView
Tmpl.listConcurrencyPoliciesSQL Text
schemaName)

-- | One prefix's concurrency pool view with live aggregates.
getConcurrencyPolicy :: (MonadArbiter m) => SchemaName -> Text -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy Text
schemaName Text
prefix =
  [ConcurrencyPolicyView] -> Maybe ConcurrencyPolicyView
forall a. [a] -> Maybe a
listToMaybe
    ([ConcurrencyPolicyView] -> Maybe ConcurrencyPolicyView)
-> m [ConcurrencyPolicyView] -> m (Maybe ConcurrencyPolicyView)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query ConcurrencyPolicyView -> m [ConcurrencyPolicyView]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query ConcurrencyPolicyView
Tmpl.getConcurrencyPolicySQL Text
schemaName Text
prefix)

-- | List a prefix's keys with effective cap and fill fraction, paginated.
listConcurrencyKeys :: (MonadArbiter m) => SchemaName -> Text -> Int -> Int -> m [ConcurrencyKeyView]
listConcurrencyKeys :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [ConcurrencyKeyView]
listConcurrencyKeys Text
schemaName Text
prefix Int
limit Int
offset =
  Query ConcurrencyKeyView -> m [ConcurrencyKeyView]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
    (Text -> Text -> Int64 -> Int64 -> Query ConcurrencyKeyView
Tmpl.listConcurrencyKeysSQL Text
schemaName Text
prefix (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
limit) (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
offset))

-- | Delete drained concurrency rows with no live job across the given tables. Returns
-- the number pruned. A key whose advisory try-lock is contended is skipped until the next pass.
pruneConcurrencyKeys :: (MonadArbiter m) => SchemaName -> [TableName] -> m Int64
pruneConcurrencyKeys :: forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
pruneConcurrencyKeys Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
pruneConcurrencyKeys Text
schemaName [Text]
tableNames = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
  dead <-
    Query Text -> m [Text]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> [Text] -> Query Text
Tmpl.lockDeadConcurrencyKeysSQL Text
schemaName [Text]
tableNames)
  if null dead
    then pure 0
    else do
      locked <- MA.executeQuery (Tmpl.tryLockDeadConcurrencyAdvisorySQL dead)
      if null locked
        then pure 0
        else MA.executeStatement (Tmpl.pruneLockedConcurrencyKeysSQL schemaName tableNames locked)

-- | Lock the count rows, then recount those keys under the lock. A key seeded after
-- the lock pass is left to its triggers.
reconcileConcurrencyCounts :: (MonadArbiter m) => SchemaName -> [TableName] -> m Int64
reconcileConcurrencyCounts :: forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
reconcileConcurrencyCounts Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
reconcileConcurrencyCounts Text
schemaName [Text]
tableNames = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
  held <- Query Text -> m [Text]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Query Text
Tmpl.lockConcurrencyCountsSQL Text
schemaName)
  rows <- MA.executeQuery (Tmpl.reconcileConcurrencyCountsSQL schemaName tableNames held)
  pure (fromMaybe 0 (listToMaybe rows))

-- | Rebuild the counts when a crash truncated the UNLOGGED table. Returns the
-- rows it recounted.
reconcileConcurrencyCountsIfStale :: (MonadArbiter m) => SchemaName -> [TableName] -> m Int64
reconcileConcurrencyCountsIfStale :: forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
reconcileConcurrencyCountsIfStale Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
reconcileConcurrencyCountsIfStale Text
schemaName [Text]
tableNames = do
  stale <- Query Bool -> m [Bool]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> [Text] -> Query Bool
Tmpl.concurrencyCountsStaleSQL Text
schemaName [Text]
tableNames)
  if or stale then reconcileConcurrencyCounts schemaName tableNames else pure 0

-- | Reconcile then prune, skipped when no concurrency key exists. Returns
-- the rows recounted and pruned.
reconcileAndPruneConcurrency :: (MonadArbiter m) => SchemaName -> [TableName] -> m Int64
reconcileAndPruneConcurrency :: forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
reconcileAndPruneConcurrency Text
schemaName [Text]
tableNames = do
  hasKeys <- Query Bool -> m [Bool]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Query Bool
Tmpl.concurrencyHasAnyKeySQL Text
schemaName)
  if not (or hasKeys)
    then pure 0
    else
      (+)
        <$> reconcileConcurrencyCounts schemaName tableNames
        <*> pruneConcurrencyKeys schemaName tableNames

-- | Insert a job, returning it with its database-generated fields. 'Nothing' when an
-- @IgnoreDuplicate@ key already exists, or a @ReplaceDuplicate@ one names a job that is
-- claimed, force-cancel flagged, or has children. Parent and rollup state come from
-- @insertJobTree@.
insertJob
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> JobWrite payload
  -> m (Maybe (JobRead payload))
insertJob :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> JobWrite payload -> m (Maybe (JobRead payload))
insertJob Text
schemaName Text
tableName JobWrite payload
job =
  m (TraceStamp payload)
forall (m :: * -> *) payload. MonadIO m => m (TraceStamp payload)
traceStamp m (TraceStamp payload)
-> (TraceStamp payload -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload))
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \TraceStamp payload
stamp -> Text
-> Text
-> TraceStamp payload
-> JobWrite payload
-> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> TraceStamp payload
-> JobWrite payload
-> m (Maybe (JobRead payload))
insertJobStamped Text
schemaName Text
tableName TraceStamp payload
stamp JobWrite payload
job

-- | 'insertJob' over a batch in one round trip, returning the jobs inserted or
-- replaced. Jobs sharing a dedup key within the batch collapse the way
-- sequential 'insertJob' calls would.
insertJobsBatch
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [JobWrite payload]
  -- ^ Jobs to insert
  -> m [JobRead payload]
insertJobsBatch :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobWrite payload] -> m [JobRead payload]
insertJobsBatch Text
_ Text
_ [] = [JobRead payload] -> m [JobRead payload]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
insertJobsBatch Text
schemaName Text
tableName [JobWrite payload]
jobs =
  m (TraceStamp payload)
forall (m :: * -> *) payload. MonadIO m => m (TraceStamp payload)
traceStamp m (TraceStamp payload)
-> (TraceStamp payload -> m [JobRead payload])
-> m [JobRead payload]
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \TraceStamp payload
stamp -> Text
-> Text
-> TraceStamp payload
-> [JobWrite payload]
-> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> TraceStamp payload
-> [JobWrite payload]
-> m [JobRead payload]
insertJobsBatchStamped Text
schemaName Text
tableName TraceStamp payload
stamp [JobWrite payload]
jobs

-- | 'insertJobsBatch' over a stamp the caller shares across its inserts.
insertJobsBatchStamped
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> TraceStamp payload
  -> [JobWrite payload]
  -> m [JobRead payload]
insertJobsBatchStamped :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> TraceStamp payload
-> [JobWrite payload]
-> m [JobRead payload]
insertJobsBatchStamped Text
_ Text
_ TraceStamp payload
_ [] = [JobRead payload] -> m [JobRead payload]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
insertJobsBatchStamped Text
schemaName Text
tableName TraceStamp payload
stamp [JobWrite payload]
jobs = do
  let batchSrc :: Query ()
batchSrc = Codec (JobWriteSource payload) (JobRead Value)
-> [JobWriteSource payload] -> Query ()
forall s a. Codec s a -> [s] -> Query ()
batchFrag (Text -> Codec (JobWriteSource payload) (JobRead Value)
forall payload.
Text -> Codec (JobWriteSource payload) (JobRead Value)
jobCodec Text
tableName) ((JobWrite payload -> JobWriteSource payload)
-> [JobWrite payload] -> [JobWriteSource payload]
forall a b. (a -> b) -> [a] -> [b]
map (TraceStamp payload -> JobWrite payload -> JobWriteSource payload
forall payload.
JobPayload payload =>
TraceStamp payload -> JobWrite payload -> JobWriteSource payload
stampedRow TraceStamp payload
stamp) ([JobWrite payload] -> [JobWrite payload]
forall payload. [JobWrite payload] -> [JobWrite payload]
dedupBatch [JobWrite payload]
jobs))
  m [JobRead payload] -> m [JobRead payload]
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m [JobRead payload] -> m [JobRead payload])
-> m [JobRead payload] -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ do
    rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query () -> Query (JobRead Value)
Tmpl.insertJobsBatchSQL Text
schemaName Text
tableName Query ()
batchSrc)
    traverse decodePayload rawJobs

-- | 'insertJobsBatch' discarding the rows, returning the count inserted.
insertJobsBatch_
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => Text
  -> Text
  -> [JobWrite payload]
  -> m Int64
insertJobsBatch_ :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobWrite payload] -> m Int64
insertJobsBatch_ Text
_ Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
insertJobsBatch_ Text
schemaName Text
tableName [JobWrite payload]
jobs = do
  stamp <- m (TraceStamp payload)
forall (m :: * -> *) payload. MonadIO m => m (TraceStamp payload)
traceStamp
  let batchSrc = Codec (JobWriteSource payload) (JobRead Value)
-> [JobWriteSource payload] -> Query ()
forall s a. Codec s a -> [s] -> Query ()
batchFrag (Text -> Codec (JobWriteSource payload) (JobRead Value)
forall payload.
Text -> Codec (JobWriteSource payload) (JobRead Value)
jobCodec Text
tableName) ((JobWrite payload -> JobWriteSource payload)
-> [JobWrite payload] -> [JobWriteSource payload]
forall a b. (a -> b) -> [a] -> [b]
map (TraceStamp payload -> JobWrite payload -> JobWriteSource payload
forall payload.
JobPayload payload =>
TraceStamp payload -> JobWrite payload -> JobWriteSource payload
stampedRow TraceStamp payload
stamp) ([JobWrite payload] -> [JobWrite payload]
forall payload. [JobWrite payload] -> [JobWrite payload]
dedupBatch [JobWrite payload]
jobs))
  withDbTransaction (MA.executeStatement (Tmpl.insertJobsBatchSQL_ schemaName tableName batchSrc))

-- | Insert a child's result, keyed by @(parent_id, child_id)@. Its foreign key cascades
-- on the parent's ack.
insertResult
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> Int64
  -- ^ Child job id
  -> Value
  -- ^ Encoded result value
  -> m Int64
insertResult :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> Int64 -> Value -> m Int64
insertResult Text
schemaName Text
tableName Int64
parentJobId Int64
childId Value
result =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Int64 -> Value -> Query ()
Tmpl.insertResultSQL Text
schemaName Text
tableName Int64
parentJobId Int64
childId Value
result)

-- | 'insertResult' for several @(parent id, child id, result)@ rows in one statement.
insertResultsBatch
  :: (MonadArbiter m) => SchemaName -> TableName -> [(Int64, Int64, Value)] -> m Int64
insertResultsBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [(Int64, Int64, Value)] -> m Int64
insertResultsBatch Text
_ Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
insertResultsBatch Text
schemaName Text
tableName [(Int64, Int64, Value)]
rows =
  let ([Int64]
parentIds, [Int64]
childIds, [Value]
results) = [(Int64, Int64, Value)] -> ([Int64], [Int64], [Value])
forall a b c. [(a, b, c)] -> ([a], [b], [c])
unzip3 [(Int64, Int64, Value)]
rows
   in Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
        (Text -> Text -> [Int64] -> [Int64] -> [Value] -> Query ()
Tmpl.insertResultsBatchSQL Text
schemaName Text
tableName [Int64]
parentIds [Int64]
childIds [Value]
results)

-- | A parent's child results, keyed by child id.
getResultsByParent
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> m (Map.Map Int64 Value)
getResultsByParent :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Map Int64 Value)
getResultsByParent Text
schemaName Text
tableName Int64
parentJobId = do
  rows <- Query (Int64, Value) -> m [(Int64, Value)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (Int64, Value)
Tmpl.getResultsByParentSQL Text
schemaName Text
tableName Int64
parentJobId)
  pure $ Map.fromList rows

-- | A parent's DLQ'd children's last errors, keyed by child id.
getDLQChildErrorsByParent
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> m (Map.Map Int64 Text)
getDLQChildErrorsByParent :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Map Int64 Text)
getDLQChildErrorsByParent Text
schemaName Text
tableName Int64
parentJobId = do
  rows <- Query (Int64, Maybe Text) -> m [(Int64, Maybe Text)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (Int64, Maybe Text)
Tmpl.getDLQChildErrorsByParentSQL Text
schemaName Text
tableName Int64
parentJobId)
  pure $ Map.fromList $ mapMaybe (\(Int64
jid, Maybe Text
mErr) -> (Int64
jid,) (Text -> (Int64, Text)) -> Maybe Text -> Maybe (Int64, Text)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text
mErr) rows

-- | Snapshot results into @parent_state@ before DLQ move.
persistParentState
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Job id
  -> Value
  -- ^ The pre-populated parent state to persist
  -> m Int64
persistParentState :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> Value -> m Int64
persistParentState Text
schemaName Text
tableName Int64
jobId Value
state =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Value -> Int64 -> Query ()
Tmpl.persistParentStateSQL Text
schemaName Text
tableName Value
state Int64
jobId)

-- | Collapse a batch's own dedup keys in input order, matching sequential 'insertJob'
-- calls. An ignore key keeps its first job, a replace key its last, and a replace
-- anywhere in a key's run wins over the ignores.
dedupBatch :: [JobWrite payload] -> [JobWrite payload]
dedupBatch :: forall payload. [JobWrite payload] -> [JobWrite payload]
dedupBatch = Seq (JobWrite payload) -> [JobWrite payload]
forall a. Seq a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList (Seq (JobWrite payload) -> [JobWrite payload])
-> ([JobWrite payload] -> Seq (JobWrite payload))
-> [JobWrite payload]
-> [JobWrite payload]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Map Text Int, Seq (JobWrite payload)) -> Seq (JobWrite payload)
forall a b. (a, b) -> b
snd ((Map Text Int, Seq (JobWrite payload)) -> Seq (JobWrite payload))
-> ([JobWrite payload] -> (Map Text Int, Seq (JobWrite payload)))
-> [JobWrite payload]
-> Seq (JobWrite payload)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ((Map Text Int, Seq (JobWrite payload))
 -> JobWrite payload -> (Map Text Int, Seq (JobWrite payload)))
-> (Map Text Int, Seq (JobWrite payload))
-> [JobWrite payload]
-> (Map Text Int, Seq (JobWrite payload))
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' (Map Text Int, Seq (JobWrite payload))
-> JobWrite payload -> (Map Text Int, Seq (JobWrite payload))
forall {payload} {key} {q} {insertedAt} {adm}.
(Map Text Int, Seq (JobRecord payload key q insertedAt adm))
-> JobRecord payload key q insertedAt adm
-> (Map Text Int, Seq (JobRecord payload key q insertedAt adm))
step (Map Text Int
forall k a. Map k a
Map.empty, Seq (JobWrite payload)
forall a. Seq a
Seq.empty)
  where
    step :: (Map Text Int, Seq (JobRecord payload key q insertedAt adm))
-> JobRecord payload key q insertedAt adm
-> (Map Text Int, Seq (JobRecord payload key q insertedAt adm))
step (!Map Text Int
seen, !Seq (JobRecord payload key q insertedAt adm)
rows) JobRecord payload key q insertedAt adm
job = case Maybe DedupKey -> Maybe Text
dedupKeyText (JobRecord payload key q insertedAt adm -> Maybe DedupKey
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey JobRecord payload key q insertedAt adm
job) of
      Maybe Text
Nothing -> (Map Text Int
seen, Seq (JobRecord payload key q insertedAt adm)
rows Seq (JobRecord payload key q insertedAt adm)
-> JobRecord payload key q insertedAt adm
-> Seq (JobRecord payload key q insertedAt adm)
forall a. Seq a -> a -> Seq a
|> JobRecord payload key q insertedAt adm
job)
      Just Text
key -> case Text -> Map Text Int -> Maybe Int
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Text
key Map Text Int
seen of
        Maybe Int
Nothing -> (Text -> Int -> Map Text Int -> Map Text Int
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert Text
key (Seq (JobRecord payload key q insertedAt adm) -> Int
forall a. Seq a -> Int
Seq.length Seq (JobRecord payload key q insertedAt adm)
rows) Map Text Int
seen, Seq (JobRecord payload key q insertedAt adm)
rows Seq (JobRecord payload key q insertedAt adm)
-> JobRecord payload key q insertedAt adm
-> Seq (JobRecord payload key q insertedAt adm)
forall a. Seq a -> a -> Seq a
|> JobRecord payload key q insertedAt adm
job)
        Just Int
idx
          | JobRecord payload key q insertedAt adm -> Bool
forall {payload} {key} {q} {insertedAt} {adm}.
JobRecord payload key q insertedAt adm -> Bool
isReplace JobRecord payload key q insertedAt adm
job -> (Map Text Int
seen, Int
-> JobRecord payload key q insertedAt adm
-> Seq (JobRecord payload key q insertedAt adm)
-> Seq (JobRecord payload key q insertedAt adm)
forall a. Int -> a -> Seq a -> Seq a
Seq.update Int
idx JobRecord payload key q insertedAt adm
job Seq (JobRecord payload key q insertedAt adm)
rows)
          | Bool
otherwise -> (Map Text Int
seen, Seq (JobRecord payload key q insertedAt adm)
rows)

    isReplace :: JobRecord payload key q insertedAt adm -> Bool
isReplace JobRecord payload key q insertedAt adm
job = case JobRecord payload key q insertedAt adm -> Maybe DedupKey
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey JobRecord payload key q insertedAt adm
job of
      Just (ReplaceDuplicate Text
_) -> Bool
True
      Maybe DedupKey
_ -> Bool
False

dedupKeyText :: Maybe DedupKey -> Maybe Text
dedupKeyText :: Maybe DedupKey -> Maybe Text
dedupKeyText = (Maybe Text, Maybe Text) -> Maybe Text
forall a b. (a, b) -> a
fst ((Maybe Text, Maybe Text) -> Maybe Text)
-> (Maybe DedupKey -> (Maybe Text, Maybe Text))
-> Maybe DedupKey
-> Maybe Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Maybe DedupKey -> (Maybe Text, Maybe Text)
dedupParts

-- | The 'Claim.ClaimAdmission' for this payload type.
claimAdmissionFor :: forall payload. (HasConcurrency payload, HasRateLimit payload) => Claim.ClaimAdmission
claimAdmissionFor :: forall payload.
(HasConcurrency payload, HasRateLimit payload) =>
ClaimAdmission
claimAdmissionFor =
  Claim.ClaimAdmission
    { admitRateLimited :: Bool
Claim.admitRateLimited = Selector Policy payload (Maybe RateLimitKey) -> Bool
forall policy payload a. Selector policy payload a -> Bool
usesAnyPolicy (forall payload. HasRateLimit payload => RateLimitFor payload
rateLimitFor @payload)
    , admitConcurrent :: Bool
Claim.admitConcurrent = Selector ConcurrencyPolicy payload (Maybe ConcurrencyKey) -> Bool
forall policy payload a. Selector policy payload a -> Bool
usesAnyPolicy (forall payload. HasConcurrency payload => ConcurrencyFor payload
concurrencyFor @payload)
    }

-- | The claimant stamped by the claim variants that take no worker id.
anonymousClaimant :: UUID
anonymousClaimant :: UUID
anonymousClaimant = Word32 -> Word32 -> Word32 -> Word32 -> UUID
UUID.fromWords Word32
0xa4b17e40 Word32
0 Word32
0 Word32
1

-- | Claim up to @maxJobs@ visible jobs, one per group. Stamps 'anonymousClaimant'.
claimNextVisibleJobs
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int
  -> NominalDiffTime
  -> m [JobRead payload]
claimNextVisibleJobs :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> NominalDiffTime -> m [JobRead payload]
claimNextVisibleJobs Text
schemaName Text
tableName Int
maxJobs NominalDiffTime
timeout =
  Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
claimJobs Text
schemaName Text
tableName Int
maxJobs NominalDiffTime
timeout UUID
anonymousClaimant

-- | 'claimNextVisibleJobs' under a given worker id.
claimNextVisibleJobsAs
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int
  -> NominalDiffTime
  -> UUID
  -> m [JobRead payload]
claimNextVisibleJobsAs :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
claimNextVisibleJobsAs Text
schemaName Text
tableName Int
maxJobs NominalDiffTime
timeout UUID
workerId =
  Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
claimJobs Text
schemaName Text
tableName Int
maxJobs NominalDiffTime
timeout UUID
workerId

claimJobs
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int
  -> NominalDiffTime
  -> UUID
  -> m [JobRead payload]
claimJobs :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
claimJobs Text
schemaName Text
tableName Int
maxJobs NominalDiffTime
timeout UUID
workerId =
  -- Batch size 1 is the single-job claim.
  ClaimSql -> Int -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
ClaimSql -> Int -> m [JobRead payload]
claimJobsCached (Proxy payload
-> Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> ClaimSql
forall payload (proxy :: * -> *).
JobPayload payload =>
proxy payload
-> Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> ClaimSql
mkClaimSql (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @payload) Text
schemaName Text
tableName Int
1 Int
0 NominalDiffTime
timeout UUID
workerId) Int
maxJobs

-- | A pool's claim statements, rendered once per capacity in @[1 .. poolSize]@.
-- 'claimSqlFor' falls back to a fresh render outside that range.
data ClaimSql = ClaimSql
  { ClaimSql -> Text
claimSqlTable :: TableName
  , ClaimSql -> Int
claimSqlBatchSize :: Int
  , ClaimSql -> UUID
claimSqlClaimant :: UUID
  -- ^ Bound as the claim's @claimed_by@. One rendered statement serves every claimant.
  , ClaimSql -> Int -> Query (JobRead Value)
claimSqlFor :: Int -> Q.Query (JobRead Value)
  }

-- | Assemble a pool's claim statements. Every input except the per-poll capacity
-- is constant for the pool's lifetime.
mkClaimSql
  :: forall payload proxy
   . (JobPayload payload)
  => proxy payload
  -> SchemaName
  -> TableName
  -> Int
  -> Int
  -> NominalDiffTime
  -> UUID
  -> ClaimSql
mkClaimSql :: forall payload (proxy :: * -> *).
JobPayload payload =>
proxy payload
-> Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> ClaimSql
mkClaimSql proxy payload
_ Text
schemaName Text
tableName Int
batchSize Int
poolSize NominalDiffTime
timeout UUID
workerId =
  let admission :: ClaimAdmission
admission = forall payload.
(HasConcurrency payload, HasRateLimit payload) =>
ClaimAdmission
claimAdmissionFor @payload
      claimant :: [SomeParam]
claimant = [Col UUID -> UUID -> SomeParam
forall a. Col a -> a -> SomeParam
pval Col UUID
CUuid UUID
workerId]
      codec :: RowCodec (JobRead Value)
codec = Text -> RowCodec (JobRead Value)
jobRowCodec Text
tableName
      render :: Int -> Query (JobRead Value)
render Int
capacity =
        Text
-> [SomeParam] -> RowCodec (JobRead Value) -> Query (JobRead Value)
forall a. Text -> [SomeParam] -> RowCodec a -> Query a
Q.Query (Text
-> Text -> ClaimAdmission -> Int -> Int -> NominalDiffTime -> Text
Claim.claimJobsBatchedSQL Text
schemaName Text
tableName ClaimAdmission
admission Int
batchSize Int
capacity NominalDiffTime
timeout) [SomeParam]
claimant RowCodec (JobRead Value)
codec
      cache :: IntMap (Query (JobRead Value))
cache = [(Int, Query (JobRead Value))] -> IntMap (Query (JobRead Value))
forall a. [(Int, a)] -> IntMap a
IntMap.fromList [(Int
capacity, Int -> Query (JobRead Value)
render Int
capacity) | Int
capacity <- [Int
1 .. Int
poolSize]]
   in ClaimSql
        { claimSqlTable :: Text
claimSqlTable = Text
tableName
        , claimSqlBatchSize :: Int
claimSqlBatchSize = Int
batchSize
        , claimSqlClaimant :: UUID
claimSqlClaimant = UUID
workerId
        , claimSqlFor :: Int -> Query (JobRead Value)
claimSqlFor = \Int
capacity -> Query (JobRead Value)
-> Int -> IntMap (Query (JobRead Value)) -> Query (JobRead Value)
forall a. a -> Int -> IntMap a -> a
IntMap.findWithDefault (Int -> Query (JobRead Value)
render Int
capacity) Int
capacity IntMap (Query (JobRead Value))
cache
        }

-- | 'claimJobs' over a prebuilt 'ClaimSql'.
claimJobsCached
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => ClaimSql
  -> Int
  -> m [JobRead payload]
claimJobsCached :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
ClaimSql -> Int -> m [JobRead payload]
claimJobsCached ClaimSql
claimSql Int
maxJobs = m [JobRead payload] -> m [JobRead payload]
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m [JobRead payload] -> m [JobRead payload])
-> m [JobRead payload] -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ do
  rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQueryPrepared (ClaimSql -> Int -> Query (JobRead Value)
claimSqlFor ClaimSql
claimSql Int
maxJobs)
  traverse decodePayload rawJobs

-- | 'claimNextVisibleJobs' claiming up to @batchSize@ jobs from each of @maxBatches@
-- groups. Stamps 'anonymousClaimant'.
claimNextVisibleJobsBatched
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int
  -> Int
  -> NominalDiffTime
  -> m [NonEmpty (JobRead payload)]
claimNextVisibleJobsBatched :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> m [NonEmpty (JobRead payload)]
claimNextVisibleJobsBatched Text
schemaName Text
tableName Int
batchSize Int
maxBatches NominalDiffTime
timeout =
  Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> m [NonEmpty (JobRead payload)]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> m [NonEmpty (JobRead payload)]
claimJobsBatched Text
schemaName Text
tableName Int
batchSize Int
maxBatches NominalDiffTime
timeout UUID
anonymousClaimant

claimJobsBatched
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int
  -> Int
  -> NominalDiffTime
  -> UUID
  -> m [NonEmpty (JobRead payload)]
claimJobsBatched :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> m [NonEmpty (JobRead payload)]
claimJobsBatched Text
schemaName Text
tableName Int
batchSize Int
maxBatches NominalDiffTime
timeout UUID
workerId =
  ClaimSql -> Int -> m [NonEmpty (JobRead payload)]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
ClaimSql -> Int -> m [NonEmpty (JobRead payload)]
claimJobsBatchedCached (Proxy payload
-> Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> ClaimSql
forall payload (proxy :: * -> *).
JobPayload payload =>
proxy payload
-> Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> ClaimSql
mkClaimSql (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @payload) Text
schemaName Text
tableName Int
batchSize Int
0 NominalDiffTime
timeout UUID
workerId) Int
maxBatches

-- | 'claimJobsBatched' over a prebuilt 'ClaimSql'.
claimJobsBatchedCached
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => ClaimSql
  -> Int
  -> m [NonEmpty (JobRead payload)]
claimJobsBatchedCached :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
ClaimSql -> Int -> m [NonEmpty (JobRead payload)]
claimJobsBatchedCached ClaimSql
claimSql Int
maxBatches
  | ClaimSql -> Int
claimSqlBatchSize ClaimSql
claimSql Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
1 = [NonEmpty (JobRead payload)] -> m [NonEmpty (JobRead payload)]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
  | Int
maxBatches Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
1 = [NonEmpty (JobRead payload)] -> m [NonEmpty (JobRead payload)]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
  | Bool
otherwise = m [NonEmpty (JobRead payload)] -> m [NonEmpty (JobRead payload)]
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m [NonEmpty (JobRead payload)] -> m [NonEmpty (JobRead payload)])
-> m [NonEmpty (JobRead payload)] -> m [NonEmpty (JobRead payload)]
forall a b. (a -> b) -> a -> b
$ do
      rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQueryPrepared (ClaimSql -> Int -> Query (JobRead Value)
claimSqlFor ClaimSql
claimSql Int
maxBatches)
      jobs <- traverse decodePayload rawJobs
      let sorted = (JobRead payload -> Maybe Text)
-> [JobRead payload] -> [JobRead payload]
forall b a. Ord b => (a -> b) -> [a] -> [a]
sortOn JobRead payload -> Maybe Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe Text
groupKey [JobRead payload]
jobs
          groups = (JobRead payload -> JobRead payload -> Bool)
-> [JobRead payload] -> [[JobRead payload]]
forall a. (a -> a -> Bool) -> [a] -> [[a]]
groupBy (\JobRead payload
jobA JobRead payload
jobB -> JobRead payload -> Maybe Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe Text
groupKey JobRead payload
jobA Maybe Text -> Maybe Text -> Bool
forall a. Eq a => a -> a -> Bool
== JobRead payload -> Maybe Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe Text
groupKey JobRead payload
jobB) [JobRead payload]
sorted
      pure $ concatMap (chunksOfNE (claimSqlBatchSize claimSql)) $ mapMaybe NE.nonEmpty groups

-- | Split a NonEmpty list into chunks of at most @size@ elements.
chunksOfNE :: Int -> NonEmpty a -> [NonEmpty a]
chunksOfNE :: forall a. Int -> NonEmpty a -> [NonEmpty a]
chunksOfNE Int
size (a
leader :| [a]
others) = [a] -> [NonEmpty a]
go (a
leader a -> [a] -> [a]
forall a. a -> [a] -> [a]
: [a]
others)
  where
    go :: [a] -> [NonEmpty a]
go [] = []
    go (a
next : [a]
more) =
      let ([a]
chunk, [a]
rest) = Int -> [a] -> ([a], [a])
forall a. Int -> [a] -> ([a], [a])
splitAt (Int
size Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1) [a]
more
       in (a
next a -> [a] -> NonEmpty a
forall a. a -> [a] -> NonEmpty a
:| [a]
chunk) NonEmpty a -> [NonEmpty a] -> [NonEmpty a]
forall a. a -> [a] -> [a]
: [a] -> [NonEmpty a]
go [a]
rest

-- | Whether acking this job tees it into the archive (positive @archiveFor@).
archivesOnAck :: JobRead payload -> Bool
archivesOnAck :: forall payload. JobRead payload -> Bool
archivesOnAck = Bool -> (Int32 -> Bool) -> Maybe Int32 -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
False (Int32 -> Int32 -> Bool
forall a. Ord a => a -> a -> Bool
> Int32
0) (Maybe Int32 -> Bool)
-> (JobRead payload -> Maybe Int32) -> JobRead payload -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. JobRead payload -> Maybe Int32
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> Maybe Int32
archiveFor

-- | Ack a completed job. Deletes a standalone one, suspends a parent whose children are
-- still running, and wakes a parent whose last child finished. A child's ack takes
-- the parent's advisory lock. Returns 1, or 0 for a job already gone.
ackJob
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> JobRead payload
  -> m Int64
ackJob :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> JobRead payload -> m Int64
ackJob Text
schemaName Text
tableName JobRead payload
job = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ Text -> Text -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> JobRead payload -> m Int64
ackJobInner Text
schemaName Text
tableName JobRead payload
job

-- | Inner ack logic, run inside the caller's transaction.
ackJobInner
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName -> TableName -> JobRead payload -> m Int64
ackJobInner :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> JobRead payload -> m Int64
ackJobInner Text
schemaName Text
tableName JobRead payload
job = do
  Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName [JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Maybe Int64
parentId JobRead payload
job]
  let jid :: Int64
jid = JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job
      cseq :: Int64
cseq = JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job
  Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0Prepared (Bool -> Text -> Text -> Int64 -> Int64 -> Query Int64
Tmpl.smartAckJobSQL (JobRead payload -> Bool
forall payload. JobRead payload -> Bool
archivesOnAck JobRead payload
job) Text
schemaName Text
tableName Int64
jid Int64
cseq)

-- | Take the advisory lock of every distinct parent named, ascending, before any
-- row lock the caller goes on to take.
lockJobParents :: (MonadArbiter m) => SchemaName -> TableName -> [Maybe Int64] -> m ()
lockJobParents :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName [Maybe Int64]
parents =
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([Int64] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Int64]
pids)
    (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ m [Maybe Text] -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
    (m [Maybe Text] -> m ()) -> m [Maybe Text] -> m ()
forall a b. (a -> b) -> a -> b
$ Query (Maybe Text) -> m [Maybe Text]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> [Int64] -> Query (Maybe Text)
advisoryXactLockManySQL (Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName) [Int64]
pids)
  where
    pids :: [Int64]
pids = Set Int64 -> [Int64]
forall a. Set a -> [a]
Set.toAscList ([Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Maybe Int64] -> [Int64]
forall a. [Maybe a] -> [a]
catMaybes [Maybe Int64]
parents))

-- | Read a job's parent and take its advisory lock, before any row lock the caller takes.
lockParentOf :: (MonadArbiter m) => SchemaName -> TableName -> Int64 -> m (Maybe Int64)
lockParentOf :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe Int64)
lockParentOf Text
schemaName Text
tableName Int64
jobId = do
  parentRows <- Query (Maybe Int64) -> m [Maybe Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (Maybe Int64)
Tmpl.getParentIdSQL Text
schemaName Text
tableName Int64
jobId)
  let mParentId = case [Maybe Int64]
parentRows of
        [Just Int64
pid] -> Int64 -> Maybe Int64
forall a. a -> Maybe a
Just Int64
pid
        [Maybe Int64]
_ -> Maybe Int64
forall a. Maybe a
Nothing
  mParentId <$ lockJobParents schemaName tableName [mParentId]

-- | Wake every distinct parent named, ascending, matching the order the locks were taken.
resumeJobParents :: (MonadArbiter m) => TreeLocks -> SchemaName -> TableName -> [Maybe Int64] -> m ()
resumeJobParents :: forall (m :: * -> *).
MonadArbiter m =>
TreeLocks -> Text -> Text -> [Maybe Int64] -> m ()
resumeJobParents TreeLocks
locks Text
schemaName Text
tableName =
  (Int64 -> m ()) -> [Int64] -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (TreeLocks -> Text -> Text -> Int64 -> m ()
forall (m :: * -> *).
MonadArbiter m =>
TreeLocks -> Text -> Text -> Int64 -> m ()
tryResumeParent TreeLocks
locks Text
schemaName Text
tableName) ([Int64] -> m ())
-> ([Maybe Int64] -> [Int64]) -> [Maybe Int64] -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Set Int64 -> [Int64]
forall a. Set a -> [a]
Set.toAscList (Set Int64 -> [Int64])
-> ([Maybe Int64] -> Set Int64) -> [Maybe Int64] -> [Int64]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Int64] -> Set Int64)
-> ([Maybe Int64] -> [Int64]) -> [Maybe Int64] -> Set Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Maybe Int64] -> [Int64]
forall a. [Maybe a] -> [a]
catMaybes

-- | Lock every job named and all of its descendants, in one descending pass.
lockJobTrees :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m ()
lockJobTrees :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m ()
lockJobTrees Text
_ Text
_ [] = () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
lockJobTrees Text
schemaName Text
tableName [Int64]
ids = m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m Int64 -> m ()) -> m Int64 -> m ()
forall a b. (a -> b) -> a -> b
$ Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 (Text -> Text -> [Int64] -> Query Int64
Tmpl.lockJobTreesSQL Text
schemaName Text
tableName [Int64]
ids)

-- | Apply 'lockJobTrees' to the complete tree of each named job. Use these locks
-- before tree cancellation.
lockJobTreesFromRoot :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m ()
lockJobTreesFromRoot :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m ()
lockJobTreesFromRoot Text
_ Text
_ [] = () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
lockJobTreesFromRoot Text
schemaName Text
tableName [Int64]
ids =
  m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m Int64 -> m ()) -> m Int64 -> m ()
forall a b. (a -> b) -> a -> b
$ Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 (Text -> Text -> [Int64] -> Query Int64
Tmpl.lockJobTreesFromRootSQL Text
schemaName Text
tableName [Int64]
ids)

-- | Wake a suspended parent when all children are done.
tryResumeParent :: (MonadArbiter m) => TreeLocks -> SchemaName -> TableName -> Int64 -> m ()
tryResumeParent :: forall (m :: * -> *).
MonadArbiter m =>
TreeLocks -> Text -> Text -> Int64 -> m ()
tryResumeParent TreeLocks
locks Text
schemaName Text
tableName Int64
pid = do
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (TreeLocks
locks TreeLocks -> TreeLocks -> Bool
forall a. Eq a => a -> a -> Bool
== TreeLocks
TakeLocks)
    (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ m [Maybe Text] -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
    (m [Maybe Text] -> m ()) -> m [Maybe Text] -> m ()
forall a b. (a -> b) -> a -> b
$ Query (Maybe Text) -> m [Maybe Text]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
      (Text -> Int64 -> Query (Maybe Text)
advisoryXactLockSQL (Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName) Int64
pid)
  m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m Int64 -> m ()) -> m Int64 -> m ()
forall a b. (a -> b) -> a -> b
$
    Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
      (Text -> Text -> Int64 -> Query ()
Tmpl.tryWakeAncestorSQL Text
schemaName Text
tableName Int64
pid)

-- | 'ackJob' over a batch in one statement, locking the distinct parents to serialize
-- with concurrent sibling acks.
ackJobsBatch
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [JobRead payload]
  -> m [Int64]
  -- ^ Ids acked (deleted or suspended). Reclaimed jobs are absent.
ackJobsBatch :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [JobRead payload] -> m [Int64]
ackJobsBatch Text
schemaName Text
tableName = m [Int64] -> m [Int64]
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m [Int64] -> m [Int64])
-> ([JobRead payload] -> m [Int64])
-> [JobRead payload]
-> m [Int64]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Text -> [JobRead payload] -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [JobRead payload] -> m [Int64]
ackJobsBatchInner Text
schemaName Text
tableName

-- | Inner batch ack, run inside the caller's transaction.
ackJobsBatchInner
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName -> TableName -> [JobRead payload] -> m [Int64]
ackJobsBatchInner :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [JobRead payload] -> m [Int64]
ackJobsBatchInner Text
_ Text
_ [] = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
ackJobsBatchInner Text
schemaName Text
tableName [JobRead payload]
jobs = do
  Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName ((JobRead payload -> Maybe Int64)
-> [JobRead payload] -> [Maybe Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Maybe Int64
parentId [JobRead payload]
jobs)
  Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQueryPrepared
    (Bool -> Text -> Text -> [Int64] -> [Int64] -> Query Int64
Tmpl.smartAckJobsBatchSQL ((JobRead payload -> Bool) -> [JobRead payload] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any JobRead payload -> Bool
forall payload. JobRead payload -> Bool
archivesOnAck [JobRead payload]
jobs) Text
schemaName Text
tableName ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey [JobRead payload]
jobs) ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq [JobRead payload]
jobs))

-- | Extend a job's visibility timeout.
setVisibilityTimeout
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> NominalDiffTime
  -- ^ Timeout in seconds
  -> JobRead payload
  -> m Int64
  -- ^ Rows updated. 0 for a row that is gone, reclaimed, or suspended.
setVisibilityTimeout :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> NominalDiffTime -> JobRead payload -> m Int64
setVisibilityTimeout Text
schemaName Text
tableName NominalDiffTime
timeout JobRead payload
job =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Double -> Int64 -> Int64 -> Query ()
Tmpl.setVisibilityTimeoutSQL Text
schemaName Text
tableName (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac NominalDiffTime
timeout) (JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job) (JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job))

-- | What a batch visibility update found for one job.
data VisibilityUpdateInfo = VisibilityUpdateInfo
  { VisibilityUpdateInfo -> Int64
vuiJobId :: JobId
  -- ^ The job this row is for.
  , VisibilityUpdateInfo -> Bool
vuiWasHeartbeated :: Bool
  -- ^ Whether the update extended this row.
  , VisibilityUpdateInfo -> Maybe Int64
vuiCurrentDbClaimSeq :: Maybe ClaimSeq
  -- ^ The row's claim token now. 'Nothing' when the row is gone.
  , VisibilityUpdateInfo -> Bool
vuiCancelRequested :: Bool
  -- ^ 'True' when a force-cancel has flagged this job.
  , VisibilityUpdateInfo -> Bool
vuiSuspended :: Bool
  -- ^ 'True' when the row is a finalizer waiting on its children.
  , VisibilityUpdateInfo -> Maybe UUID
vuiClaimedBy :: Maybe UUID
  -- ^ Who holds the row's claim now.
  }
  deriving stock (VisibilityUpdateInfo -> VisibilityUpdateInfo -> Bool
(VisibilityUpdateInfo -> VisibilityUpdateInfo -> Bool)
-> (VisibilityUpdateInfo -> VisibilityUpdateInfo -> Bool)
-> Eq VisibilityUpdateInfo
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: VisibilityUpdateInfo -> VisibilityUpdateInfo -> Bool
== :: VisibilityUpdateInfo -> VisibilityUpdateInfo -> Bool
$c/= :: VisibilityUpdateInfo -> VisibilityUpdateInfo -> Bool
/= :: VisibilityUpdateInfo -> VisibilityUpdateInfo -> Bool
Eq, (forall x. VisibilityUpdateInfo -> Rep VisibilityUpdateInfo x)
-> (forall x. Rep VisibilityUpdateInfo x -> VisibilityUpdateInfo)
-> Generic VisibilityUpdateInfo
forall x. Rep VisibilityUpdateInfo x -> VisibilityUpdateInfo
forall x. VisibilityUpdateInfo -> Rep VisibilityUpdateInfo x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. VisibilityUpdateInfo -> Rep VisibilityUpdateInfo x
from :: forall x. VisibilityUpdateInfo -> Rep VisibilityUpdateInfo x
$cto :: forall x. Rep VisibilityUpdateInfo x -> VisibilityUpdateInfo
to :: forall x. Rep VisibilityUpdateInfo x -> VisibilityUpdateInfo
Generic, Int -> VisibilityUpdateInfo -> ShowS
[VisibilityUpdateInfo] -> ShowS
VisibilityUpdateInfo -> String
(Int -> VisibilityUpdateInfo -> ShowS)
-> (VisibilityUpdateInfo -> String)
-> ([VisibilityUpdateInfo] -> ShowS)
-> Show VisibilityUpdateInfo
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> VisibilityUpdateInfo -> ShowS
showsPrec :: Int -> VisibilityUpdateInfo -> ShowS
$cshow :: VisibilityUpdateInfo -> String
show :: VisibilityUpdateInfo -> String
$cshowList :: [VisibilityUpdateInfo] -> ShowS
showList :: [VisibilityUpdateInfo] -> ShowS
Show)

-- | 'setVisibilityTimeout' over a batch, reporting each row through 'VisibilityUpdateInfo'.
setVisibilityTimeoutBatch
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> NominalDiffTime
  -- ^ Timeout in seconds
  -> [JobRead payload]
  -> m [VisibilityUpdateInfo]
  -- ^ One status record per job targeted.
setVisibilityTimeoutBatch :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> NominalDiffTime
-> [JobRead payload]
-> m [VisibilityUpdateInfo]
setVisibilityTimeoutBatch Text
_ Text
_ NominalDiffTime
_ [] = [VisibilityUpdateInfo] -> m [VisibilityUpdateInfo]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
setVisibilityTimeoutBatch Text
schemaName Text
tableName NominalDiffTime
timeout [JobRead payload]
jobs = do
  let valuesFrag :: Query ()
valuesFrag =
        Text -> [Query ()] -> Query ()
Q.sepBy
          Text
","
          [ [QQ.sql|(#{jid :: CInt8}, #{cseq :: CInt8}, #{holder :: Maybe CUuid})|]
          | JobRead payload
job <- [JobRead payload]
jobs
          , let jid :: Int64
jid = JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job
          , let cseq :: Int64
cseq = JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job
          , let holder :: Maybe UUID
holder = JobRead payload -> Maybe UUID
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Maybe UUID
claimedBy JobRead payload
job
          ]

  Query VisibilityUpdateInfo -> m [VisibilityUpdateInfo]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
    (Query VisibilityUpdateInfo -> m [VisibilityUpdateInfo])
-> Query VisibilityUpdateInfo -> m [VisibilityUpdateInfo]
forall a b. (a -> b) -> a -> b
$ RowCodec VisibilityUpdateInfo
-> Query () -> Query VisibilityUpdateInfo
forall a. RowCodec a -> Query () -> Query a
Q.rows RowCodec VisibilityUpdateInfo
visibilityUpdateCodec
    (Query () -> Query VisibilityUpdateInfo)
-> Query () -> Query VisibilityUpdateInfo
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Query () -> [Int64] -> Double -> Query ()
Tmpl.setVisibilityTimeoutBatchSQL Text
schemaName Text
tableName Query ()
valuesFrag ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey [JobRead payload]
jobs) (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac NominalDiffTime
timeout)

-- | Park a failed job for its retry backoff, recording the error. Returns 0 for a job
-- another worker holds.
updateJobForRetry
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> NominalDiffTime
  -- ^ Backoff timeout in seconds
  -> Text
  -- ^ Error message
  -> JobRead payload
  -> m Int64
updateJobForRetry :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> NominalDiffTime -> Text -> JobRead payload -> m Int64
updateJobForRetry Text
schemaName Text
tableName NominalDiffTime
backoff Text
errorMsg JobRead payload
job =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Text -> Int64 -> Int64 -> Query ()
Tmpl.updateJobForRetrySQL Text
schemaName Text
tableName (NominalDiffTime -> Int64
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling NominalDiffTime
backoff) Text
errorMsg (JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job) (JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job))

-- | Soft-nack a job. Hands back the attempt the claim consumed and records no failure.
-- Returns 0 for a job another worker holds.
nackJob
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> JobRead payload
  -> m Int64
nackJob :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> JobRead payload -> m Int64
nackJob Text
schemaName Text
tableName JobRead payload
job =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Int64 -> Int32 -> Query ()
Tmpl.nackJobSQL Text
schemaName Text
tableName (JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job) (JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job) (JobRead payload -> Int32
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int32
attempts JobRead payload
job))

-- | 'nackJob' over a batch in one statement, returning the ids nacked. Jobs another
-- worker holds are absent.
nackJobsBatch
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [JobRead payload]
  -> m [Int64]
nackJobsBatch :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [JobRead payload] -> m [Int64]
nackJobsBatch Text
_ Text
_ [] = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
nackJobsBatch Text
schemaName Text
tableName [JobRead payload]
jobs =
  Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
    (Text -> Text -> [Int64] -> [Int64] -> [Int32] -> Query Int64
Tmpl.nackJobsBatchSQL Text
schemaName Text
tableName ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey [JobRead payload]
jobs) ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq [JobRead payload]
jobs) ((JobRead payload -> Int32) -> [JobRead payload] -> [Int32]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int32
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int32
attempts [JobRead payload]
jobs))

-- | Whether the caller already took the parent and tree locks over its whole set.
data TreeLocks
  = TakeLocks
  | LocksHeld
  deriving stock (TreeLocks -> TreeLocks -> Bool
(TreeLocks -> TreeLocks -> Bool)
-> (TreeLocks -> TreeLocks -> Bool) -> Eq TreeLocks
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: TreeLocks -> TreeLocks -> Bool
== :: TreeLocks -> TreeLocks -> Bool
$c/= :: TreeLocks -> TreeLocks -> Bool
/= :: TreeLocks -> TreeLocks -> Bool
Eq, Int -> TreeLocks -> ShowS
[TreeLocks] -> ShowS
TreeLocks -> String
(Int -> TreeLocks -> ShowS)
-> (TreeLocks -> String)
-> ([TreeLocks] -> ShowS)
-> Show TreeLocks
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> TreeLocks -> ShowS
showsPrec :: Int -> TreeLocks -> ShowS
$cshow :: TreeLocks -> String
show :: TreeLocks -> String
$cshowList :: [TreeLocks] -> ShowS
showList :: [TreeLocks] -> ShowS
Show)

-- | Move a job to the DLQ, cascading a rollup parent's descendants with it and waking
-- the parent of a child. Returns 0 for a job another worker holds.
moveToDLQ
  :: forall m payload
   . (MonadArbiter m)
  => TreeLocks
  -> SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Text
  -- ^ Error message (the final error that caused the DLQ move)
  -> JobRead payload
  -> m Int64
moveToDLQ :: forall (m :: * -> *) payload.
MonadArbiter m =>
TreeLocks -> Text -> Text -> Text -> JobRead payload -> m Int64
moveToDLQ TreeLocks
locks Text
schemaName Text
tableName Text
errorMsg JobRead payload
job =
  TreeLocks
-> DLQMove
-> Text
-> Text
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> m Int64
forall (m :: * -> *).
MonadArbiter m =>
TreeLocks
-> DLQMove
-> Text
-> Text
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> m Int64
moveToDLQFields
    TreeLocks
locks
    DLQMove
Tmpl.MoveNow
    Text
schemaName
    Text
tableName
    Text
errorMsg
    (JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job)
    (JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job)
    (JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Maybe Int64
parentId JobRead payload
job)
    (JobRead payload -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
isRollup JobRead payload
job)

-- | 'moveToDLQ' driven by scalar fields, for callers without a typed 'JobRead'.
moveToDLQFields
  :: (MonadArbiter m)
  => TreeLocks
  -> Tmpl.DLQMove
  -> SchemaName
  -> TableName
  -> Text
  -- ^ Error message (the final error that caused the DLQ move)
  -> Int64
  -- ^ Job id
  -> Int64
  -- ^ Claim token (for the optimistic move check)
  -> Maybe Int64
  -- ^ Parent id, if a child
  -> Bool
  -- ^ Whether the job is a rollup finalizer
  -> m Int64
moveToDLQFields :: forall (m :: * -> *).
MonadArbiter m =>
TreeLocks
-> DLQMove
-> Text
-> Text
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> m Int64
moveToDLQFields TreeLocks
locks DLQMove
move Text
schemaName Text
tableName Text
errorMsg Int64
jobId Int64
cseq Maybe Int64
mParentId Bool
rollup = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (TreeLocks
locks TreeLocks -> TreeLocks -> Bool
forall a. Eq a => a -> a -> Bool
== TreeLocks
TakeLocks) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName [Maybe Int64
mParentId]
    Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
rollup (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> [Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m ()
lockJobTrees Text
schemaName Text
tableName [Int64
jobId]
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
rollup (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Int64 -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m ()
snapshotTreeRollups Text
schemaName Text
tableName Int64
jobId
  rows <- Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 (DLQMove -> Text -> Text -> Int64 -> Int64 -> Text -> Query Int64
Tmpl.moveToDLQSQL DLQMove
move Text
schemaName Text
tableName Int64
jobId Int64
cseq Text
errorMsg)
  when (rows > 0) $ do
    when rollup $ void $ cascadeChildrenToDLQ schemaName tableName jobId "Parent moved to DLQ"
    for_ mParentId $ \Int64
pid ->
      TreeLocks -> Text -> Text -> Int64 -> m ()
forall (m :: * -> *).
MonadArbiter m =>
TreeLocks -> Text -> Text -> Int64 -> m ()
tryResumeParent TreeLocks
LocksHeld Text
schemaName Text
tableName Int64
pid
  pure rows

-- | Move every descendant of a rollup parent to the DLQ under one error message.
-- Returns the number moved.
cascadeChildrenToDLQ
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> Text
  -- ^ Error message for cascaded children
  -> m Int64
cascadeChildrenToDLQ :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> Text -> m Int64
cascadeChildrenToDLQ Text
schemaName Text
tableName Int64
parentJobId Text
errorMsg =
  Text -> Query Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict
    Text
"cascadeChildrenToDLQ"
    (Text -> Text -> Int64 -> Text -> Query Int64
Tmpl.cascadeChildrenToDLQSQL Text
schemaName Text
tableName Int64
parentJobId Text
errorMsg)

-- | Snapshot child results for every rollup finalizer in a job's tree, the job
-- included. Persists accumulated results into @parent_state@.
snapshotTreeRollups
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Root of the tree being moved
  -> m ()
snapshotTreeRollups :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m ()
snapshotTreeRollups Text
schemaName Text
tableName Int64
parentJobId = do
  rollupIds <- Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query Int64
Tmpl.treeRollupIdsSQL Text
schemaName Text
tableName Int64
parentJobId)
  for_ rollupIds $ \Int64
rid -> do
    (results, errors, snap, _) <- Text
-> Text
-> Int64
-> m (Map Int64 Value, Map Int64 Text, Maybe Value, Map Int64 Text)
forall (m :: * -> *).
MonadArbiter m =>
Text
-> Text
-> Int64
-> m (Map Int64 Value, Map Int64 Text, Maybe Value, Map Int64 Text)
readChildResultsRaw Text
schemaName Text
tableName Int64
rid
    let merged = Map Int64 Value
-> Map Int64 Text -> Maybe Value -> Map Int64 (Either Text Value)
mergeRawChildResults Map Int64 Value
results Map Int64 Text
errors Maybe Value
snap
    unless (Map.null merged)
      $ void
      $ persistParentState schemaName tableName rid (toJSON merged)

-- | 'moveToDLQ' over a batch, each job under its own error message. Jobs another worker
-- reclaimed are skipped. Returns the number moved.
moveToDLQBatch
  :: forall m payload
   . (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [(JobRead payload, Text)]
  -- ^ List of (job, error message) pairs
  -> m Int64
moveToDLQBatch :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [(JobRead payload, Text)] -> m Int64
moveToDLQBatch Text
_ Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
moveToDLQBatch Text
schemaName Text
tableName [(JobRead payload, Text)]
jobsWithErrors = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
  let ids :: [Int64]
ids = ((JobRead payload, Text) -> Int64)
-> [(JobRead payload, Text)] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey (JobRead payload -> Int64)
-> ((JobRead payload, Text) -> JobRead payload)
-> (JobRead payload, Text)
-> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead payload, Text) -> JobRead payload
forall a b. (a, b) -> a
fst) [(JobRead payload, Text)]
jobsWithErrors
      cseqs :: [Int64]
cseqs = ((JobRead payload, Text) -> Int64)
-> [(JobRead payload, Text)] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead payload -> Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq (JobRead payload -> Int64)
-> ((JobRead payload, Text) -> JobRead payload)
-> (JobRead payload, Text)
-> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead payload, Text) -> JobRead payload
forall a b. (a, b) -> a
fst) [(JobRead payload, Text)]
jobsWithErrors
      msgs :: [Text]
msgs = ((JobRead payload, Text) -> Text)
-> [(JobRead payload, Text)] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead payload, Text) -> Text
forall a b. (a, b) -> b
snd [(JobRead payload, Text)]
jobsWithErrors
      rollupIds :: [Int64]
rollupIds = Set Int64 -> [Int64]
forall a. Set a -> [a]
Set.toList (Set Int64 -> [Int64])
-> ([Int64] -> Set Int64) -> [Int64] -> [Int64]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Int64] -> [Int64]) -> [Int64] -> [Int64]
forall a b. (a -> b) -> a -> b
$ ((JobRead payload, Text) -> Int64)
-> [(JobRead payload, Text)] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey (JobRead payload -> Int64)
-> ((JobRead payload, Text) -> JobRead payload)
-> (JobRead payload, Text)
-> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead payload, Text) -> JobRead payload
forall a b. (a, b) -> a
fst) (((JobRead payload, Text) -> Bool)
-> [(JobRead payload, Text)] -> [(JobRead payload, Text)]
forall a. (a -> Bool) -> [a] -> [a]
filter (JobRead payload -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
isRollup (JobRead payload -> Bool)
-> ((JobRead payload, Text) -> JobRead payload)
-> (JobRead payload, Text)
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead payload, Text) -> JobRead payload
forall a b. (a, b) -> a
fst) [(JobRead payload, Text)]
jobsWithErrors)
  Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName (((JobRead payload, Text) -> Maybe Int64)
-> [(JobRead payload, Text)] -> [Maybe Int64]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Maybe Int64
parentId (JobRead payload -> Maybe Int64)
-> ((JobRead payload, Text) -> JobRead payload)
-> (JobRead payload, Text)
-> Maybe Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead payload, Text) -> JobRead payload
forall a b. (a, b) -> a
fst) [(JobRead payload, Text)]
jobsWithErrors)
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([Int64] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Int64]
rollupIds) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    -- Every row the move will lock, plus the trees, in one descending pass.
    Text -> Text -> [Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m ()
lockJobTrees Text
schemaName Text
tableName [Int64]
ids
    -- Before the move, which takes a named rollup's results with it.
    [Int64] -> (Int64 -> m ()) -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
t a -> (a -> f b) -> f ()
for_ [Int64]
rollupIds ((Int64 -> m ()) -> m ()) -> (Int64 -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Int64 -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m ()
snapshotTreeRollups Text
schemaName Text
tableName
  moved <- [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Int64] -> Set Int64) -> m [Int64] -> m (Set Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> [Int64] -> [Int64] -> [Text] -> Query Int64
Tmpl.moveToDLQBatchSQL Text
schemaName Text
tableName [Int64]
ids [Int64]
cseqs [Text]
msgs)
  let movedJobs = ((JobRead payload, Text) -> Bool)
-> [(JobRead payload, Text)] -> [(JobRead payload, Text)]
forall a. (a -> Bool) -> [a] -> [a]
filter ((Int64 -> Set Int64 -> Bool) -> Set Int64 -> Int64 -> Bool
forall a b c. (a -> b -> c) -> b -> a -> c
flip Int64 -> Set Int64 -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member Set Int64
moved (Int64 -> Bool)
-> ((JobRead payload, Text) -> Int64)
-> (JobRead payload, Text)
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey (JobRead payload -> Int64)
-> ((JobRead payload, Text) -> JobRead payload)
-> (JobRead payload, Text)
-> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead payload, Text) -> JobRead payload
forall a b. (a, b) -> a
fst) [(JobRead payload, Text)]
jobsWithErrors
  for_ movedJobs $ \(JobRead payload
job, Text
_) ->
    Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (JobRead payload -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
isRollup JobRead payload
job)
      (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void
      (m Int64 -> m ()) -> m Int64 -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Int64 -> Text -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> Text -> m Int64
cascadeChildrenToDLQ Text
schemaName Text
tableName (JobRead payload -> Int64
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job) Text
"Parent moved to DLQ"
  resumeJobParents LocksHeld schemaName tableName (map (parentId . fst) movedJobs)
  pure (fromIntegral (Set.size moved))

-- ---------------------------------------------------------------------------
-- Dead Letter Queue Operations
-- ---------------------------------------------------------------------------

-- | Retry a job from the DLQ, re-inserting it with a fresh attempt count. The dedup key
-- is left behind.
retryFromDLQ
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ DLQ job id
  -> m (Maybe (JobRead payload))
retryFromDLQ :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (JobRead payload))
retryFromDLQ Text
schemaName Text
tableName Int64
dlqId = m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ do
  rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (JobRead Value)
Tmpl.retryFromDLQSQL Text
schemaName Text
tableName Int64
dlqId)
  traverse decodePayload (listToMaybe rawJobs)

-- | Whether a DLQ job with the given id exists.
dlqJobExists
  :: (MonadArbiter m)
  => Text
  -> Text
  -> Int64
  -> m Bool
dlqJobExists :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Bool
dlqJobExists Text
schemaName Text
tableName Int64
dlqId = do
  rows <- Query Bool -> m [Bool]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query Bool
Tmpl.dlqJobExistsSQL Text
schemaName Text
tableName Int64
dlqId)
  pure (fromMaybe False (listToMaybe rows))

-- ---------------------------------------------------------------------------
-- Filtered Query Operations
-- ---------------------------------------------------------------------------

-- | List filtered jobs under an explicit sort spec. @Nothing@ for both sort arguments
-- orders by @id DESC@.
listJobsFilteredOrdered
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Tmpl.JobFilter]
  -- ^ Composable filters
  -> Maybe Tmpl.JobSortColumn
  -- ^ Sort column (defaults to 'Tmpl.JsId')
  -> Maybe Tmpl.SortDir
  -- ^ Sort direction (defaults to 'Tmpl.SortDesc')
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
listJobsFilteredOrdered :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [JobRead payload]
listJobsFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset
  | (JobFilter -> Bool) -> [JobFilter] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any JobFilter -> Bool
isStatusFilter [JobFilter]
filters =
      ((JobRead payload, JobStatus) -> JobRead payload)
-> [(JobRead payload, JobStatus)] -> [JobRead payload]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead payload, JobStatus) -> JobRead payload
forall a b. (a, b) -> a
fst ([(JobRead payload, JobStatus)] -> [JobRead payload])
-> m [(JobRead payload, JobStatus)] -> m [JobRead payload]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead payload, JobStatus)]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead payload, JobStatus)]
listJobsWithStatus Text
schemaName Text
tableName [JobFilter]
filters Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset
  | Bool
otherwise = do
      let orderBy :: Text
orderBy = Maybe JobSortColumn -> Maybe SortDir -> Text
Tmpl.buildJobsOrderBy Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir
      rawJobs <-
        Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Query (JobRead Value) -> m [JobRead Value])
-> Query (JobRead Value) -> m [JobRead Value]
forall a b. (a -> b) -> a -> b
$
          Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (JobRead Value)
Tmpl.listJobsFilteredSQL
            Text
schemaName
            Text
tableName
            ([JobFilter] -> Query ()
buildWhereClause [JobFilter]
filters)
            Text
orderBy
            (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
limit)
            (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
offset)
      traverse decodePayload rawJobs

isStatusFilter :: Tmpl.JobFilter -> Bool
isStatusFilter :: JobFilter -> Bool
isStatusFilter (Tmpl.FilterStatus JobStatus
_) = Bool
True
isStatusFilter JobFilter
_ = Bool
False

-- | Decode a job row and its derived @status@ column. Validate the status after
-- the backend codec runs. An unknown SQL value causes a parsing failure.
jobRowWithStatusCodec :: TableName -> RowCodec (JobRead Value, Text)
jobRowWithStatusCodec :: Text -> RowCodec (JobRead Value, Text)
jobRowWithStatusCodec Text
tableName =
  (,) (JobRead Value -> Text -> (JobRead Value, Text))
-> RowCodec (JobRead Value)
-> Ap NullCol (Text -> (JobRead Value, Text))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> RowCodec (JobRead Value)
jobRowCodec Text
tableName Ap NullCol (Text -> (JobRead Value, Text))
-> Ap NullCol Text -> RowCodec (JobRead Value, Text)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Text -> Ap NullCol Text
forall a. Text -> Col a -> RowCodec a
col Text
"status" Col Text
CText

-- | 'listJobsFilteredOrdered' that also returns each job's derived status.
listJobsWithStatus
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> [Tmpl.JobFilter]
  -> Maybe Tmpl.JobSortColumn
  -> Maybe Tmpl.SortDir
  -> Int
  -> Int
  -> m [(JobRead payload, JobStatus)]
listJobsWithStatus :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead payload, JobStatus)]
listJobsWithStatus Text
schemaName Text
tableName [JobFilter]
filters Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset = do
  let orderBy :: Text
orderBy = Maybe JobSortColumn -> Maybe SortDir -> Text
Tmpl.buildJobsOrderBy Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir
      query :: Query ()
query =
        Text -> Text -> Query () -> Text -> Int64 -> Int64 -> Query ()
Tmpl.listJobsWithStatusSQL
          Text
schemaName
          Text
tableName
          ([JobFilter] -> Query ()
buildWhereClause [JobFilter]
filters)
          Text
orderBy
          (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
limit)
          (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
offset)
  rows <- Query (JobRead Value, Text) -> m [(JobRead Value, Text)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (RowCodec (JobRead Value, Text)
-> Query () -> Query (JobRead Value, Text)
forall a. RowCodec a -> Query () -> Query a
Q.rows (Text -> RowCodec (JobRead Value, Text)
jobRowWithStatusCodec Text
tableName) Query ()
query)
  traverse decodeJobStatusRow rows

-- | Decode the payload and strictly validate the SQL-derived status.
decodeJobStatusRow
  :: (JobPayload payload, MonadArbiter m)
  => (JobRead Value, Text)
  -> m (JobRead payload, JobStatus)
decodeJobStatusRow :: forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
(JobRead Value, Text) -> m (JobRead payload, JobStatus)
decodeJobStatusRow (JobRead Value
job, Text
rawStatus) = do
  decodedJob <- JobRead Value -> m (JobRead payload)
forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
JobRead Value -> m (JobRead payload)
decodePayload JobRead Value
job
  status <- either throwParsing pure (jobStatusFromText rawStatus)
  pure (decodedJob, status)

-- | List filtered jobs, newest first.
listJobsFiltered
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Tmpl.JobFilter]
  -- ^ Composable filters
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
listJobsFiltered :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
listJobsFiltered Text
schemaName Text
tableName [JobFilter]
filters =
  Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [JobRead payload]
listJobsFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe JobSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing

-- | Count jobs with composable filters.
countJobsFiltered
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Tmpl.JobFilter]
  -- ^ Composable filters
  -> m Int64
countJobsFiltered :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countJobsFiltered Text
schemaName Text
tableName [JobFilter]
filters = do
  Text -> Query Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict Text
"countJobsFiltered" (Text -> Text -> Query () -> Query Int64
Tmpl.countJobsFilteredSQL Text
schemaName Text
tableName ([JobFilter] -> Query ()
buildWhereClause [JobFilter]
filters))

-- | List filtered DLQ jobs under an explicit sort spec. @Nothing@ for both sort
-- arguments orders by @failed_at DESC@.
listDLQFilteredOrdered
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Tmpl.JobFilter]
  -- ^ Composable filters
  -> Maybe Tmpl.DLQSortColumn
  -- ^ Sort column (defaults to 'Tmpl.DlqFailedAt')
  -> Maybe Tmpl.SortDir
  -- ^ Sort direction (defaults to 'Tmpl.SortDesc')
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [DLQ.DLQJob payload]
listDLQFilteredOrdered :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob payload]
listDLQFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe DLQSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset = do
  let orderBy :: Text
orderBy = Maybe DLQSortColumn -> Maybe SortDir -> Text
Tmpl.buildDLQOrderBy Maybe DLQSortColumn
mSortBy Maybe SortDir
mSortDir
  rawRows <-
    Query (Int64, UTCTime, JobRead Value)
-> m [(Int64, UTCTime, JobRead Value)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Query (Int64, UTCTime, JobRead Value)
 -> m [(Int64, UTCTime, JobRead Value)])
-> Query (Int64, UTCTime, JobRead Value)
-> m [(Int64, UTCTime, JobRead Value)]
forall a b. (a -> b) -> a -> b
$
      Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (Int64, UTCTime, JobRead Value)
Tmpl.listDLQFilteredSQL
        Text
schemaName
        Text
tableName
        ([JobFilter] -> Query ()
buildWhereClause [JobFilter]
filters)
        Text
orderBy
        (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
limit)
        (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
offset)
  traverse decodeDLQRow rawRows

-- | List filtered DLQ jobs, most recently failed first.
listDLQFiltered
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Tmpl.JobFilter]
  -- ^ Composable filters
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [DLQ.DLQJob payload]
listDLQFiltered :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [DLQJob payload]
listDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters =
  Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob payload]
listDLQFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe DLQSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing

-- | Count DLQ jobs with composable filters.
countDLQFiltered
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Tmpl.JobFilter]
  -- ^ Composable filters
  -> m Int64
countDLQFiltered :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters =
  Text -> Query Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict Text
"countDLQFiltered" (Text -> Text -> Query () -> Query Int64
Tmpl.countDLQFilteredSQL Text
schemaName Text
tableName ([JobFilter] -> Query ()
buildWhereClause [JobFilter]
filters))

decodeDLQRow
  :: (JobPayload payload, MonadArbiter m)
  => (Int64, UTCTime, JobRead Value)
  -> m (DLQ.DLQJob payload)
decodeDLQRow :: forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
(Int64, UTCTime, JobRead Value) -> m (DLQJob payload)
decodeDLQRow (Int64
dlqId, UTCTime
dlqFailedAt, JobRead Value
rawJob) = do
  jobSnapshot <- JobRead Value -> m (JobRead payload)
forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
JobRead Value -> m (JobRead payload)
decodePayload JobRead Value
rawJob
  pure $
    DLQ.DLQJob
      { DLQ.dlqPrimaryKey = dlqId
      , DLQ.failedAt = dlqFailedAt
      , DLQ.jobSnapshot = jobSnapshot
      }

-- | List archived (completed) jobs with composable filters and a typed sort
-- (defaulting to most recent first).
listArchiveFiltered
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> [Tmpl.JobFilter]
  -> Maybe Tmpl.ArchiveSortColumn
  -> Maybe Tmpl.SortDir
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [Archive.ArchiveJob payload]
listArchiveFiltered :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob payload]
listArchiveFiltered Text
schemaName Text
tableName [JobFilter]
filters Maybe ArchiveSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset = do
  let orderBy :: Text
orderBy = Maybe ArchiveSortColumn -> Maybe SortDir -> Text
Tmpl.buildArchiveOrderBy Maybe ArchiveSortColumn
mSortBy Maybe SortDir
mSortDir
  rawRows <-
    Query (Int64, UTCTime, JobRead Value, Maybe Value)
-> m [(Int64, UTCTime, JobRead Value, Maybe Value)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Query (Int64, UTCTime, JobRead Value, Maybe Value)
 -> m [(Int64, UTCTime, JobRead Value, Maybe Value)])
-> Query (Int64, UTCTime, JobRead Value, Maybe Value)
-> m [(Int64, UTCTime, JobRead Value, Maybe Value)]
forall a b. (a -> b) -> a -> b
$
      Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (Int64, UTCTime, JobRead Value, Maybe Value)
Tmpl.listArchiveFilteredSQL
        Text
schemaName
        Text
tableName
        ([JobFilter] -> Query ()
buildWhereClause [JobFilter]
filters)
        Text
orderBy
        (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
limit)
        (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
offset)
  traverse decodeArchiveRow rawRows

-- | List archived jobs (most recent first).
listArchiveJobs
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int
  -> Int
  -> m [Archive.ArchiveJob payload]
listArchiveJobs :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> Int -> m [ArchiveJob payload]
listArchiveJobs Text
schemaName Text
tableName = Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob payload]
listArchiveFiltered Text
schemaName Text
tableName [] Maybe ArchiveSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing

-- | Fetch a single archived job by its original job id.
getArchivedJobById
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int64
  -> m (Maybe (Archive.ArchiveJob payload))
getArchivedJobById :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (ArchiveJob payload))
getArchivedJobById Text
schemaName Text
tableName Int64
jobId =
  [ArchiveJob payload] -> Maybe (ArchiveJob payload)
forall a. [a] -> Maybe a
listToMaybe
    ([ArchiveJob payload] -> Maybe (ArchiveJob payload))
-> m [ArchiveJob payload] -> m (Maybe (ArchiveJob payload))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob payload]
listArchiveFiltered Text
schemaName Text
tableName [Int64 -> JobFilter
Tmpl.FilterJobId Int64
jobId] Maybe ArchiveSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing Int
1 Int
0

-- | Delete one archived job by its archive primary key. Returns rows deleted.
deleteArchiveJob :: (MonadArbiter m) => SchemaName -> TableName -> Int64 -> m Int64
deleteArchiveJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
deleteArchiveJob Text
schemaName Text
tableName Int64
archiveId =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Text -> Int64 -> Query ()
Tmpl.deleteArchiveJobSQL Text
schemaName Text
tableName Int64
archiveId)

-- | Delete archived jobs by archive primary key. Returns rows deleted.
deleteArchiveJobsBatch :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m Int64
deleteArchiveJobsBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
deleteArchiveJobsBatch Text
_ Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
deleteArchiveJobsBatch Text
schemaName Text
tableName [Int64]
archiveIds =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Text -> [Int64] -> Query ()
Tmpl.deleteArchiveJobsBatchSQL Text
schemaName Text
tableName [Int64]
archiveIds)

-- | Re-enqueue an archived job as a fresh standalone job, keeping the archive
-- row. Returns the new job, or @Nothing@ when the archive row no longer exists.
reEnqueueFromArchive
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName -> TableName -> Int64 -> m (Maybe (JobRead payload))
reEnqueueFromArchive :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (JobRead payload))
reEnqueueFromArchive Text
schemaName Text
tableName Int64
archiveId = m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ do
  rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (JobRead Value)
Tmpl.reEnqueueFromArchiveSQL Text
schemaName Text
tableName Int64
archiveId)
  traverse decodePayload (listToMaybe rawJobs)

-- | Store a completed root job's result on its archive row. No-ops when the job
-- was not archived. Returns rows updated.
updateArchiveResult
  :: (MonadArbiter m) => SchemaName -> TableName -> Int64 -> Value -> m Int64
updateArchiveResult :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> Value -> m Int64
updateArchiveResult Text
schemaName Text
tableName Int64
jobId Value
result =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Value -> Int64 -> Query ()
Tmpl.updateArchiveResultSQL Text
schemaName Text
tableName Value
result Int64
jobId)

-- | 'updateArchiveResult' for several @(job id, result)@ pairs in one statement.
updateArchiveResultsBatch
  :: (MonadArbiter m) => SchemaName -> TableName -> [(Int64, Value)] -> m Int64
updateArchiveResultsBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [(Int64, Value)] -> m Int64
updateArchiveResultsBatch Text
_ Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
updateArchiveResultsBatch Text
schemaName Text
tableName [(Int64, Value)]
pairs =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (([Int64] -> [Value] -> Query ()) -> ([Int64], [Value]) -> Query ()
forall a b c. (a -> b -> c) -> (a, b) -> c
uncurry (Text -> Text -> [Int64] -> [Value] -> Query ()
Tmpl.updateArchiveResultsBatchSQL Text
schemaName Text
tableName) ([(Int64, Value)] -> ([Int64], [Value])
forall a b. [(a, b)] -> ([a], [b])
unzip [(Int64, Value)]
pairs))

-- | List archived jobs in a group, most recent first.
listArchivedJobsByGroupKey
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Text
  -> Int
  -> Int
  -> m [Archive.ArchiveJob payload]
listArchivedJobsByGroupKey :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Text -> Int -> Int -> m [ArchiveJob payload]
listArchivedJobsByGroupKey Text
schemaName Text
tableName Text
key =
  Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob payload]
listArchiveFiltered Text
schemaName Text
tableName [Text -> JobFilter
Tmpl.FilterGroupKey Text
key] Maybe ArchiveSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing

-- | Count archived jobs with composable filters.
countArchiveFiltered
  :: (MonadArbiter m)
  => SchemaName
  -> TableName
  -> [Tmpl.JobFilter]
  -> m Int64
countArchiveFiltered :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countArchiveFiltered Text
schemaName Text
tableName [JobFilter]
filters =
  Text -> Query Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict Text
"countArchiveFiltered" (Text -> Text -> Query () -> Query Int64
Tmpl.countArchiveFilteredSQL Text
schemaName Text
tableName ([JobFilter] -> Query ()
buildWhereClause [JobFilter]
filters))

decodeArchiveRow
  :: (JobPayload payload, MonadArbiter m)
  => (Int64, UTCTime, JobRead Value, Maybe Value)
  -> m (Archive.ArchiveJob payload)
decodeArchiveRow :: forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
(Int64, UTCTime, JobRead Value, Maybe Value)
-> m (ArchiveJob payload)
decodeArchiveRow (Int64
aId, UTCTime
aCompletedAt, JobRead Value
rawJob, Maybe Value
aResult) = do
  snapshot <- JobRead Value -> m (JobRead payload)
forall payload (m :: * -> *).
(JobPayload payload, MonadArbiter m) =>
JobRead Value -> m (JobRead payload)
decodePayload JobRead Value
rawJob
  pure $
    Archive.ArchiveJob
      { Archive.archivePrimaryKey = aId
      , Archive.completedAt = aCompletedAt
      , Archive.jobSnapshot = snapshot
      , Archive.archivedResult = aResult
      }

-- | Purge expired archived jobs across all queues. Each row uses its
-- @archive_expires_at@ value. Return the total rows purged and queues with
-- errors. Run one small batch at each reaper tick.
purgeArchives
  :: (MonadArbiter m)
  => SchemaName -> [TableName] -> m (Int64, [Text])
purgeArchives :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
purgeArchives =
  (Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadUnliftIO m =>
(Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
sweepQueues ((Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text]))
-> (Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
queue ->
    m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Query ()
Q.raw (Text -> Text -> Text
Tmpl.purgeArchiveSQL Text
schemaName Text
queue)))

-- | List DLQ jobs, most recently failed first.
listDLQJobs
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [DLQ.DLQJob payload]
listDLQJobs :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> Int -> m [DLQJob payload]
listDLQJobs Text
schemaName Text
tableName = Text -> Text -> [JobFilter] -> Int -> Int -> m [DLQJob payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [DLQJob payload]
listDLQFiltered Text
schemaName Text
tableName []

-- | List a parent's DLQ'd children, most recently failed first.
listDLQJobsByParent
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [DLQ.DLQJob payload]
listDLQJobsByParent :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> Int -> Int -> m [DLQJob payload]
listDLQJobsByParent Text
schemaName Text
tableName Int64
parentJobId =
  Text -> Text -> [JobFilter] -> Int -> Int -> m [DLQJob payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [DLQJob payload]
listDLQFiltered Text
schemaName Text
tableName [Int64 -> JobFilter
Tmpl.FilterParentId Int64
parentJobId]

-- | Count DLQ jobs matching a parent_id.
countDLQJobsByParent
  :: (MonadArbiter m)
  => SchemaName -> TableName -> Int64 -> m Int64
countDLQJobsByParent :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
countDLQJobsByParent Text
schemaName Text
tableName Int64
parentJobId =
  Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countDLQFiltered Text
schemaName Text
tableName [Int64 -> JobFilter
Tmpl.FilterParentId Int64
parentJobId]

-- | Delete a DLQ job, resuming its parent when no sibling is left.
deleteDLQJob
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ DLQ job id
  -> m Int64
deleteDLQJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
deleteDLQJob Text
schemaName Text
tableName Int64
dlqId = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
  rows <- Query (Maybe Int64) -> m [Maybe Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (Maybe Int64)
Tmpl.deleteDLQJobSQL Text
schemaName Text
tableName Int64
dlqId)
  case rows of
    [] -> Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
    (Just Int64
pid : [Maybe Int64]
_) -> do
      TreeLocks -> Text -> Text -> Int64 -> m ()
forall (m :: * -> *).
MonadArbiter m =>
TreeLocks -> Text -> Text -> Int64 -> m ()
tryResumeParent TreeLocks
TakeLocks Text
schemaName Text
tableName Int64
pid
      Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
1
    [Maybe Int64]
_ -> Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
1

-- | Delete jobs by id via the given query builder, then resume any parents left
-- childless. The query must return each deleted row's id and parent_id. Returns
-- the ids deleted.
deleteJobsResumingParents
  :: (MonadArbiter m)
  => SchemaName
  -> TableName
  -> ([Int64] -> Q.Query (Int64, Maybe Int64))
  -> [Int64]
  -> m [Int64]
deleteJobsResumingParents :: forall (m :: * -> *).
MonadArbiter m =>
Text
-> Text
-> ([Int64] -> Query (Int64, Maybe Int64))
-> [Int64]
-> m [Int64]
deleteJobsResumingParents Text
_ Text
_ [Int64] -> Query (Int64, Maybe Int64)
_ [] = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
deleteJobsResumingParents Text
schemaName Text
tableName [Int64] -> Query (Int64, Maybe Int64)
mkSql [Int64]
jobIds = m [Int64] -> m [Int64]
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m [Int64] -> m [Int64]) -> m [Int64] -> m [Int64]
forall a b. (a -> b) -> a -> b
$ do
  rows <- Query (Int64, Maybe Int64) -> m [(Int64, Maybe Int64)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery ([Int64] -> Query (Int64, Maybe Int64)
mkSql [Int64]
jobIds)
  resumeJobParents TakeLocks schemaName tableName (map snd rows)
  pure (map fst rows)

-- | Delete multiple jobs from the dead letter queue, resuming any parents left
-- childless. Returns the total number of DLQ jobs deleted.
deleteDLQJobsBatch
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Int64]
  -- ^ DLQ job ids
  -> m Int64
deleteDLQJobsBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
deleteDLQJobsBatch Text
schemaName Text
tableName [Int64]
dlqIds =
  Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Int64) -> ([Int64] -> Int) -> [Int64] -> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int64] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length
    ([Int64] -> Int64) -> m [Int64] -> m Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text
-> Text
-> ([Int64] -> Query (Int64, Maybe Int64))
-> [Int64]
-> m [Int64]
forall (m :: * -> *).
MonadArbiter m =>
Text
-> Text
-> ([Int64] -> Query (Int64, Maybe Int64))
-> [Int64]
-> m [Int64]
deleteJobsResumingParents Text
schemaName Text
tableName (Text -> Text -> [Int64] -> Query (Int64, Maybe Int64)
Tmpl.deleteDLQJobsBatchSQL Text
schemaName Text
tableName) [Int64]
dlqIds

-- | Delete force-cancel-flagged jobs @owner@ holds or no live lease holds, resuming
-- any parents left childless. Returns the ids it deleted.
deleteCancelledJobs
  :: (MonadArbiter m)
  => SchemaName
  -> TableName
  -> Maybe UUID
  -> [Int64]
  -> m [Int64]
deleteCancelledJobs :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe UUID -> [Int64] -> m [Int64]
deleteCancelledJobs Text
schemaName Text
tableName Maybe UUID
owner =
  Text
-> Text
-> ([Int64] -> Query (Int64, Maybe Int64))
-> [Int64]
-> m [Int64]
forall (m :: * -> *).
MonadArbiter m =>
Text
-> Text
-> ([Int64] -> Query (Int64, Maybe Int64))
-> [Int64]
-> m [Int64]
deleteJobsResumingParents Text
schemaName Text
tableName (Text -> Text -> Maybe UUID -> [Int64] -> Query (Int64, Maybe Int64)
Tmpl.deleteCancelledJobsSQL Text
schemaName Text
tableName Maybe UUID
owner)

-- ---------------------------------------------------------------------------
-- Admin Operations
-- ---------------------------------------------------------------------------

-- | List jobs, newest first.
listJobs
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
listJobs :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> Int -> m [JobRead payload]
listJobs Text
schemaName Text
tableName = Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
listJobsFiltered Text
schemaName Text
tableName []

-- | Whether a job with the given id exists in the table, without decoding it.
jobExists :: (MonadArbiter m) => SchemaName -> TableName -> Int64 -> m Bool
jobExists :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Bool
jobExists Text
schemaName Text
tableName Int64
jobId =
  [Bool] -> Bool
forall (t :: * -> *). Foldable t => t Bool -> Bool
or ([Bool] -> Bool) -> m [Bool] -> m Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query Bool -> m [Bool]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query Bool
Tmpl.jobExistsSQL Text
schemaName Text
tableName Int64
jobId)

-- | Fetch a job by id.
getJobById
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Job id
  -> m (Maybe (JobRead payload))
getJobById :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (JobRead payload))
getJobById Text
schemaName Text
tableName Int64
jobId = do
  rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (JobRead Value)
Tmpl.getJobByIdSQL Text
schemaName Text
tableName Int64
jobId)
  traverse decodePayload (listToMaybe rawJobs)

-- | 'getJobById' that also returns the job's derived status.
getJobByIdWithStatus
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int64
  -> m (Maybe (JobRead payload, JobStatus))
getJobByIdWithStatus :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (JobRead payload, JobStatus))
getJobByIdWithStatus Text
schemaName Text
tableName Int64
jobId = do
  rows <-
    Query (JobRead Value, Text) -> m [(JobRead Value, Text)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Query (JobRead Value, Text) -> m [(JobRead Value, Text)])
-> Query (JobRead Value, Text) -> m [(JobRead Value, Text)]
forall a b. (a -> b) -> a -> b
$
      RowCodec (JobRead Value, Text)
-> Query () -> Query (JobRead Value, Text)
forall a. RowCodec a -> Query () -> Query a
Q.rows (Text -> RowCodec (JobRead Value, Text)
jobRowWithStatusCodec Text
tableName) (Text -> Text -> Int64 -> Query ()
Tmpl.getJobByIdWithStatusSQL Text
schemaName Text
tableName Int64
jobId)
  traverse decodeJobStatusRow (listToMaybe rows)

-- | Get a single job by its dedup key.
getJobByDedupKey
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -> TableName
  -> Text
  -> m (Maybe (JobRead payload))
getJobByDedupKey :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Text -> m (Maybe (JobRead payload))
getJobByDedupKey Text
schemaName Text
tableName Text
key = do
  rawJobs <- Query (JobRead Value) -> m [JobRead Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Text -> Query (JobRead Value)
Tmpl.getJobByDedupKeySQL Text
schemaName Text
tableName Text
key)
  traverse decodePayload (listToMaybe rawJobs)

-- | Get all jobs for a specific group key.
getJobsByGroup
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Text
  -- ^ Group key
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
getJobsByGroup :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Text -> Int -> Int -> m [JobRead payload]
getJobsByGroup Text
schemaName Text
tableName Text
key =
  Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
listJobsFiltered Text
schemaName Text
tableName [Text -> JobFilter
Tmpl.FilterGroupKey Text
key]

-- | Delete a job by id. Returns 0 for a job with children, which 'cancelJobCascade'
-- takes. A deleted child with no siblings left resumes its parent for a
-- completion round.
cancelJob
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Job id
  -> m Int64
cancelJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
cancelJob Text
schemaName Text
tableName Int64
jobId = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
cancelJobInner Text
schemaName Text
tableName Int64
jobId

-- | Inner cancel logic, run inside the caller's transaction.
cancelJobInner
  :: (MonadArbiter m)
  => SchemaName -> TableName -> Int64 -> m Int64
cancelJobInner :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
cancelJobInner Text
schemaName Text
tableName Int64
jobId = do
  m (Maybe Int64) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (Maybe Int64) -> m ()) -> m (Maybe Int64) -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Int64 -> m (Maybe Int64)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe Int64)
lockParentOf Text
schemaName Text
tableName Int64
jobId
  Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 (Text -> Text -> Int64 -> Query Int64
Tmpl.cancelJobSQL Text
schemaName Text
tableName Int64
jobId)

-- | 'cancelJob' over several ids in one transaction. The last sibling cancelled finds
-- the parent childless and resumes it. Locks the union of parents and rows first.
-- Returns the number deleted.
cancelJobsBatch
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Int64]
  -- ^ Job ids
  -> m Int64
cancelJobsBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
cancelJobsBatch Text
_ Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
cancelJobsBatch Text
schemaName Text
tableName [Int64]
jobIds =
  m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
    let ids :: [Int64]
ids = Set Int64 -> [Int64]
forall a. Set a -> [a]
Set.toList ([Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList [Int64]
jobIds)
    parents <- Query (Maybe Int64) -> m [Maybe Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> [Int64] -> Query (Maybe Int64)
Tmpl.getParentIdsSQL Text
schemaName Text
tableName [Int64]
ids)
    lockJobParents schemaName tableName parents
    lockJobTrees schemaName tableName ids
    sum <$> traverse (countOr0 . Tmpl.cancelJobSQL schemaName tableName) ids

-- | Make a delayed or retrying job immediately visible. Refuses an in-flight job.
promoteJob
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Job id
  -> m Int64
  -- ^ Number of rows updated
promoteJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
promoteJob Text
schemaName Text
tableName Int64
jobId =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Query ()
Tmpl.promoteJobSQL Text
schemaName Text
tableName Int64
jobId)

-- | Per-status breakdown of a queue. The per-status counts partition the queue
-- and sum to 'totalJobs', mirroring the derived job status taxonomy.
data QueueStats = QueueStats
  { QueueStats -> Int64
totalJobs :: Int64
  -- ^ Total number of jobs in the queue
  , QueueStats -> Int64
readyJobs :: Int64
  -- ^ Jobs claimable right now (visible and unleased)
  , QueueStats -> Int64
inFlightJobs :: Int64
  -- ^ Jobs currently leased by a worker (a retry attempt in progress)
  , QueueStats -> Int64
scheduledJobs :: Int64
  -- ^ Jobs delayed until a future @not_visible_until@ (never yet attempted)
  , QueueStats -> Int64
backoffJobs :: Int64
  -- ^ Unclaimed jobs with an attempt spent, waiting out a delay
  , QueueStats -> Int64
throttledJobs :: Int64
  -- ^ Jobs parked by a rate limit until tokens refill
  , QueueStats -> Int64
suspendedJobs :: Int64
  -- ^ Suspended jobs (e.g. rollup finalizers awaiting their children)
  , QueueStats -> Int64
cancelledJobs :: Int64
  -- ^ Force-cancelled jobs flagged for teardown and awaiting the reaper
  , QueueStats -> Maybe Double
oldestReadyAgeSeconds :: Maybe Double
  -- ^ Age in seconds of the oldest @ready@ job (Nothing when none are ready).
  , QueueStats -> Maybe Double
oldestInFlightAgeSeconds :: Maybe Double
  -- ^ Seconds since the oldest in-flight job was claimed (Nothing when none are
  -- leased). Measures work still running.
  , QueueStats -> Int64
dlqJobs :: Int64
  -- ^ Entries in the companion DLQ table. Outside the status partition and 'totalJobs'.
  , QueueStats -> Map Text Int64
kindCounts :: Map Text Int64
  -- ^ Depth by declared payload variant. Rows with no declared label are left out.
  }
  deriving stock (QueueStats -> QueueStats -> Bool
(QueueStats -> QueueStats -> Bool)
-> (QueueStats -> QueueStats -> Bool) -> Eq QueueStats
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: QueueStats -> QueueStats -> Bool
== :: QueueStats -> QueueStats -> Bool
$c/= :: QueueStats -> QueueStats -> Bool
/= :: QueueStats -> QueueStats -> Bool
Eq, (forall x. QueueStats -> Rep QueueStats x)
-> (forall x. Rep QueueStats x -> QueueStats) -> Generic QueueStats
forall x. Rep QueueStats x -> QueueStats
forall x. QueueStats -> Rep QueueStats x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. QueueStats -> Rep QueueStats x
from :: forall x. QueueStats -> Rep QueueStats x
$cto :: forall x. Rep QueueStats x -> QueueStats
to :: forall x. Rep QueueStats x -> QueueStats
Generic, Int -> QueueStats -> ShowS
[QueueStats] -> ShowS
QueueStats -> String
(Int -> QueueStats -> ShowS)
-> (QueueStats -> String)
-> ([QueueStats] -> ShowS)
-> Show QueueStats
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> QueueStats -> ShowS
showsPrec :: Int -> QueueStats -> ShowS
$cshow :: QueueStats -> String
show :: QueueStats -> String
$cshowList :: [QueueStats] -> ShowS
showList :: [QueueStats] -> ShowS
Show)

instance ToJSON QueueStats where
  toJSON :: QueueStats -> Value
toJSON QueueStats
stats =
    [Pair] -> Value
object
      [ Key
"totalJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
totalJobs QueueStats
stats
      , Key
"readyJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
readyJobs QueueStats
stats
      , Key
"inFlightJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
inFlightJobs QueueStats
stats
      , Key
"scheduledJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
scheduledJobs QueueStats
stats
      , Key
"backoffJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
backoffJobs QueueStats
stats
      , Key
"throttledJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
throttledJobs QueueStats
stats
      , Key
"suspendedJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
suspendedJobs QueueStats
stats
      , Key
"cancelledJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
cancelledJobs QueueStats
stats
      , Key
"oldestReadyAgeSeconds" Key -> Maybe Double -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Maybe Double
oldestReadyAgeSeconds QueueStats
stats
      , Key
"oldestInFlightAgeSeconds" Key -> Maybe Double -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Maybe Double
oldestInFlightAgeSeconds QueueStats
stats
      , Key
"dlqJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
dlqJobs QueueStats
stats
      , Key
"kindCounts" Key -> Map Text Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Map Text Int64
kindCounts QueueStats
stats
      ]

instance FromJSON QueueStats where
  parseJSON :: Value -> Parser QueueStats
parseJSON = String
-> (Object -> Parser QueueStats) -> Value -> Parser QueueStats
forall a. String -> (Object -> Parser a) -> Value -> Parser a
withObject String
"QueueStats" ((Object -> Parser QueueStats) -> Value -> Parser QueueStats)
-> (Object -> Parser QueueStats) -> Value -> Parser QueueStats
forall a b. (a -> b) -> a -> b
$ \Object
obj ->
    Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> QueueStats
QueueStats
      (Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Maybe Double
 -> Maybe Double
 -> Int64
 -> Map Text Int64
 -> QueueStats)
-> Parser Int64
-> Parser
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"totalJobs"
      Parser
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Parser Int64
-> Parser
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"readyJobs"
      Parser
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Parser Int64
-> Parser
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"inFlightJobs"
      Parser
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Parser Int64
-> Parser
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"scheduledJobs"
      Parser
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Parser Int64
-> Parser
     (Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"backoffJobs"
      Parser
  (Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Parser Int64
-> Parser
     (Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"throttledJobs"
      Parser
  (Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Parser Int64
-> Parser
     (Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"suspendedJobs"
      Parser
  (Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Parser Int64
-> Parser
     (Maybe Double
      -> Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"cancelledJobs"
      Parser
  (Maybe Double
   -> Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
-> Parser (Maybe Double)
-> Parser (Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser (Maybe Double)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"oldestReadyAgeSeconds"
      Parser (Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
-> Parser (Maybe Double)
-> Parser (Int64 -> Map Text Int64 -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser (Maybe Double)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"oldestInFlightAgeSeconds"
      Parser (Int64 -> Map Text Int64 -> QueueStats)
-> Parser Int64 -> Parser (Map Text Int64 -> QueueStats)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"dlqJobs"
      Parser (Map Text Int64 -> QueueStats)
-> Parser (Map Text Int64) -> Parser QueueStats
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser (Maybe (Map Text Int64))
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"kindCounts" Parser (Maybe (Map Text Int64))
-> Map Text Int64 -> Parser (Map Text Int64)
forall a. Parser (Maybe a) -> a -> Parser a
.!= Map Text Int64
forall k a. Map k a
Map.empty

-- | All-zero counts, the fallback for an aggregate query that returned no row.
emptyQueueStats :: QueueStats
emptyQueueStats :: QueueStats
emptyQueueStats = Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> QueueStats
QueueStats Int64
0 Int64
0 Int64
0 Int64
0 Int64
0 Int64
0 Int64
0 Int64
0 Maybe Double
forall a. Maybe a
Nothing Maybe Double
forall a. Maybe a
Nothing Int64
0 Map Text Int64
forall k a. Map k a
Map.empty

-- | The per-status depths a 'QueueStats' carries.
queueStatusCounts :: QueueStats -> [(JobStatus, Int64)]
queueStatusCounts :: QueueStats -> [(JobStatus, Int64)]
queueStatusCounts QueueStats
stats =
  [ (JobStatus
Ready, QueueStats -> Int64
readyJobs QueueStats
stats)
  , (JobStatus
InFlight, QueueStats -> Int64
inFlightJobs QueueStats
stats)
  , (JobStatus
Scheduled, QueueStats -> Int64
scheduledJobs QueueStats
stats)
  , (JobStatus
Backoff, QueueStats -> Int64
backoffJobs QueueStats
stats)
  , (JobStatus
Throttled, QueueStats -> Int64
throttledJobs QueueStats
stats)
  , (JobStatus
Suspended, QueueStats -> Int64
suspendedJobs QueueStats
stats)
  , (JobStatus
Cancelled, QueueStats -> Int64
cancelledJobs QueueStats
stats)
  ]

-- | Decodes the single aggregate row produced by 'Tmpl.getQueueStatsSQL', whose
-- select list is built from these same columns.
statsRowCodec :: RowCodec QueueStats
statsRowCodec :: RowCodec QueueStats
statsRowCodec =
  Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> QueueStats
QueueStats
    (Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Int64
 -> Maybe Double
 -> Maybe Double
 -> Int64
 -> Map Text Int64
 -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"total_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"ready_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"in_flight_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Int64
      -> Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"scheduled_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Int64
   -> Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Int64
      -> Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"backoff_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Int64
   -> Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Int64
      -> Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"throttled_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Int64
   -> Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Int64
      -> Maybe Double
      -> Maybe Double
      -> Int64
      -> Map Text Int64
      -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"suspended_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Int64
   -> Maybe Double
   -> Maybe Double
   -> Int64
   -> Map Text Int64
   -> QueueStats)
-> Ap NullCol Int64
-> Ap
     NullCol
     (Maybe Double
      -> Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"cancelled_jobs" Col Int64
CInt8
    Ap
  NullCol
  (Maybe Double
   -> Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
-> Ap NullCol (Maybe Double)
-> Ap
     NullCol (Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Double -> Ap NullCol (Maybe Double)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"oldest_ready_age_seconds" Col Double
CFloat8
    Ap NullCol (Maybe Double -> Int64 -> Map Text Int64 -> QueueStats)
-> Ap NullCol (Maybe Double)
-> Ap NullCol (Int64 -> Map Text Int64 -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Double -> Ap NullCol (Maybe Double)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"oldest_in_flight_age_seconds" Col Double
CFloat8
    Ap NullCol (Int64 -> Map Text Int64 -> QueueStats)
-> Ap NullCol Int64 -> Ap NullCol (Map Text Int64 -> QueueStats)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"dlq_jobs" Col Int64
CInt8
    Ap NullCol (Map Text Int64 -> QueueStats)
-> Ap NullCol (Map Text Int64) -> RowCodec QueueStats
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (Maybe Value -> Map Text Int64
decodeKindCounts (Maybe Value -> Map Text Int64)
-> Ap NullCol (Maybe Value) -> Ap NullCol (Map Text Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Col Value -> Ap NullCol (Maybe Value)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"kind_counts" Col Value
CJsonb)

-- | The @jsonb_object_agg@ rollup as a map.
decodeKindCounts :: Maybe Value -> Map Text Int64
decodeKindCounts :: Maybe Value -> Map Text Int64
decodeKindCounts = (Value -> Map Text Int64) -> Maybe Value -> Map Text Int64
forall m a. Monoid m => (a -> m) -> Maybe a -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap (Result (Map Text Int64) -> Map Text Int64
forall {k} {a}. Result (Map k a) -> Map k a
fromResult (Result (Map Text Int64) -> Map Text Int64)
-> (Value -> Result (Map Text Int64)) -> Value -> Map Text Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Value -> Result (Map Text Int64)
forall a. FromJSON a => Value -> Result a
fromJSON)
  where
    fromResult :: Result (Map k a) -> Map k a
fromResult = \case
      Success Map k a
counts -> Map k a
counts
      Error String
_ -> Map k a
forall k a. Map k a
Map.empty

-- | A queue's per-status counts and backlog ages.
getQueueStats
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> [Text]
  -- ^ The labels the payload declares. 'kindCounts' covers these labels.
  -> m QueueStats
getQueueStats :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Text] -> m QueueStats
getQueueStats Text
schemaName Text
tableName [Text]
kinds = do
  rows <- Query QueueStats -> m [QueueStats]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (RowCodec QueueStats -> Text -> Text -> [Text] -> Query QueueStats
forall a. RowCodec a -> Text -> Text -> [Text] -> Query a
Tmpl.getQueueStatsSQL RowCodec QueueStats
statsRowCodec Text
schemaName Text
tableName [Text]
kinds)

  -- The aggregate query returns one row. The empty fallback covers an unexpected
  -- truncation.
  pure (fromMaybe emptyQueueStats (listToMaybe rows))

-- | A landing-overview row: a queue's stats plus its pause state.
data QueueOverview = QueueOverview
  { QueueOverview -> Text
overviewQueue :: Text
  , QueueOverview -> QueueStats
overviewStats :: QueueStats
  , QueueOverview -> Bool
overviewQueuePaused :: Bool
  , QueueOverview -> Int64
overviewWorkersLive :: Int64
  , QueueOverview -> Int64
overviewWorkersPaused :: Int64
  }
  deriving stock (QueueOverview -> QueueOverview -> Bool
(QueueOverview -> QueueOverview -> Bool)
-> (QueueOverview -> QueueOverview -> Bool) -> Eq QueueOverview
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: QueueOverview -> QueueOverview -> Bool
== :: QueueOverview -> QueueOverview -> Bool
$c/= :: QueueOverview -> QueueOverview -> Bool
/= :: QueueOverview -> QueueOverview -> Bool
Eq, (forall x. QueueOverview -> Rep QueueOverview x)
-> (forall x. Rep QueueOverview x -> QueueOverview)
-> Generic QueueOverview
forall x. Rep QueueOverview x -> QueueOverview
forall x. QueueOverview -> Rep QueueOverview x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. QueueOverview -> Rep QueueOverview x
from :: forall x. QueueOverview -> Rep QueueOverview x
$cto :: forall x. Rep QueueOverview x -> QueueOverview
to :: forall x. Rep QueueOverview x -> QueueOverview
Generic, Int -> QueueOverview -> ShowS
[QueueOverview] -> ShowS
QueueOverview -> String
(Int -> QueueOverview -> ShowS)
-> (QueueOverview -> String)
-> ([QueueOverview] -> ShowS)
-> Show QueueOverview
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> QueueOverview -> ShowS
showsPrec :: Int -> QueueOverview -> ShowS
$cshow :: QueueOverview -> String
show :: QueueOverview -> String
$cshowList :: [QueueOverview] -> ShowS
showList :: [QueueOverview] -> ShowS
Show)

instance ToJSON QueueOverview where
  toJSON :: QueueOverview -> Value
toJSON QueueOverview
overview =
    [Pair] -> Value
object
      [ Key
"queue" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueOverview -> Text
overviewQueue QueueOverview
overview
      , Key
"stats" Key -> QueueStats -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueOverview -> QueueStats
overviewStats QueueOverview
overview
      , Key
"paused" Key -> Bool -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueOverview -> Bool
overviewQueuePaused QueueOverview
overview
      , Key
"workersLive" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueOverview -> Int64
overviewWorkersLive QueueOverview
overview
      , Key
"workersPaused" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueOverview -> Int64
overviewWorkersPaused QueueOverview
overview
      ]

instance FromJSON QueueOverview where
  parseJSON :: Value -> Parser QueueOverview
parseJSON = String
-> (Object -> Parser QueueOverview)
-> Value
-> Parser QueueOverview
forall a. String -> (Object -> Parser a) -> Value -> Parser a
withObject String
"QueueOverview" ((Object -> Parser QueueOverview) -> Value -> Parser QueueOverview)
-> (Object -> Parser QueueOverview)
-> Value
-> Parser QueueOverview
forall a b. (a -> b) -> a -> b
$ \Object
obj ->
    Text -> QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview
QueueOverview
      (Text -> QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview)
-> Parser Text
-> Parser (QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Object
obj Object -> Key -> Parser Text
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"queue"
      Parser (QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview)
-> Parser QueueStats
-> Parser (Bool -> Int64 -> Int64 -> QueueOverview)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser QueueStats
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"stats"
      Parser (Bool -> Int64 -> Int64 -> QueueOverview)
-> Parser Bool -> Parser (Int64 -> Int64 -> QueueOverview)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Bool
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"paused"
      Parser (Int64 -> Int64 -> QueueOverview)
-> Parser Int64 -> Parser (Int64 -> QueueOverview)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"workersLive"
      Parser (Int64 -> QueueOverview)
-> Parser Int64 -> Parser QueueOverview
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser Int64
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"workersPaused"

allStatsRowCodec :: RowCodec QueueOverview
allStatsRowCodec :: RowCodec QueueOverview
allStatsRowCodec =
  Text -> QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview
QueueOverview
    (Text -> QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview)
-> Ap NullCol Text
-> Ap
     NullCol (QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Col Text -> Ap NullCol Text
forall a. Text -> Col a -> RowCodec a
col Text
"queue" Col Text
CText
    Ap NullCol (QueueStats -> Bool -> Int64 -> Int64 -> QueueOverview)
-> RowCodec QueueStats
-> Ap NullCol (Bool -> Int64 -> Int64 -> QueueOverview)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> RowCodec QueueStats
statsRowCodec
    Ap NullCol (Bool -> Int64 -> Int64 -> QueueOverview)
-> Ap NullCol Bool -> Ap NullCol (Int64 -> Int64 -> QueueOverview)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Bool -> Ap NullCol Bool
forall a. Text -> Col a -> RowCodec a
col Text
"queue_paused" Col Bool
CBool
    Ap NullCol (Int64 -> Int64 -> QueueOverview)
-> Ap NullCol Int64 -> Ap NullCol (Int64 -> QueueOverview)
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"workers_live" Col Int64
CInt8
    Ap NullCol (Int64 -> QueueOverview)
-> Ap NullCol Int64 -> RowCodec QueueOverview
forall a b. Ap NullCol (a -> b) -> Ap NullCol a -> Ap NullCol b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> Col Int64 -> Ap NullCol Int64
forall a. Text -> Col a -> RowCodec a
col Text
"workers_paused" Col Int64
CInt8

-- | Every queue's stats plus pause state in one query, for the landing overview.
getAllQueueStats
  :: (MonadArbiter m)
  => SchemaName
  -> [(TableName, [Text])]
  -- ^ Each queue with the labels its payload declares.
  -> m [QueueOverview]
getAllQueueStats :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [(Text, [Text])] -> m [QueueOverview]
getAllQueueStats Text
_ [] = [QueueOverview] -> m [QueueOverview]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
getAllQueueStats Text
schemaName [(Text, [Text])]
queueKinds =
  Query QueueOverview -> m [QueueOverview]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (RowCodec QueueOverview
-> Text -> [(Text, [Text])] -> Query QueueOverview
forall a. RowCodec a -> Text -> [(Text, [Text])] -> Query a
Tmpl.allQueueStatsSQL RowCodec QueueOverview
allStatsRowCodec Text
schemaName [(Text, [Text])]
queueKinds)

-- ---------------------------------------------------------------------------
-- Count Operations
-- ---------------------------------------------------------------------------

-- | Count every job in a table.
countJobs
  :: (MonadArbiter m)
  => SchemaName -> TableName -> m Int64
countJobs :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
countJobs Text
schemaName Text
tableName = Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countJobsFiltered Text
schemaName Text
tableName []

-- | Count a group's jobs.
countJobsByGroup
  :: (MonadArbiter m)
  => SchemaName -> TableName -> Text -> m Int64
countJobsByGroup :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> m Int64
countJobsByGroup Text
schemaName Text
tableName Text
key =
  Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countJobsFiltered Text
schemaName Text
tableName [Text -> JobFilter
Tmpl.FilterGroupKey Text
key]

-- | Count a queue's DLQ jobs.
countDLQJobs
  :: (MonadArbiter m)
  => SchemaName -> TableName -> m Int64
countDLQJobs :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
countDLQJobs Text
schemaName Text
tableName = Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countDLQFiltered Text
schemaName Text
tableName []

-- ---------------------------------------------------------------------------
-- Parent-Child Operations
-- ---------------------------------------------------------------------------

-- | List jobs filtered by parent_id with pagination.
getJobsByParent
  :: forall m payload
   . (JobPayload payload, MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent id
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
getJobsByParent :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> Int -> Int -> m [JobRead payload]
getJobsByParent Text
schemaName Text
tableName Int64
pid =
  Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
listJobsFiltered Text
schemaName Text
tableName [Int64 -> JobFilter
Tmpl.FilterParentId Int64
pid]

-- | Count jobs matching a parent_id.
countJobsByParent
  :: (MonadArbiter m)
  => SchemaName -> TableName -> Int64 -> m Int64
countJobsByParent :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
countJobsByParent Text
schemaName Text
tableName Int64
pid =
  Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countJobsFiltered Text
schemaName Text
tableName [Int64 -> JobFilter
Tmpl.FilterParentId Int64
pid]

-- | Child counts as @(total, paused)@ per parent id, over a batch. Parents with none
-- are absent.
countChildrenBatch
  :: (MonadArbiter m)
  => SchemaName -> TableName -> [Int64] -> m (Map.Map Int64 (Int64, Int64))
countChildrenBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m (Map Int64 (Int64, Int64))
countChildrenBatch Text
_ Text
_ [] = Map Int64 (Int64, Int64) -> m (Map Int64 (Int64, Int64))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Map Int64 (Int64, Int64)
forall k a. Map k a
Map.empty
countChildrenBatch Text
schemaName Text
tableName [Int64]
ids = do
  rows <-
    Query (Int64, (Int64, Int64)) -> m [(Int64, (Int64, Int64))]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Query (Int64, (Int64, Int64)) -> m [(Int64, (Int64, Int64))])
-> Query (Int64, (Int64, Int64)) -> m [(Int64, (Int64, Int64))]
forall a b. (a -> b) -> a -> b
$
      RowCodec (Int64, (Int64, Int64))
-> Query () -> Query (Int64, (Int64, Int64))
forall a. RowCodec a -> Query () -> Query a
Q.rows RowCodec (Int64, (Int64, Int64))
parentCountCodec (Text -> Text -> [Int64] -> Query ()
Tmpl.countChildrenBatchSQL Text
schemaName Text
tableName [Int64]
ids)
  pure $ Map.fromList rows

-- | DLQ child counts per parent id, over a batch. Parents with none are absent.
countDLQChildrenBatch
  :: (MonadArbiter m)
  => SchemaName -> TableName -> [Int64] -> m (Map.Map Int64 Int64)
countDLQChildrenBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m (Map Int64 Int64)
countDLQChildrenBatch Text
_ Text
_ [] = Map Int64 Int64 -> m (Map Int64 Int64)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Map Int64 Int64
forall k a. Map k a
Map.empty
countDLQChildrenBatch Text
schemaName Text
tableName [Int64]
ids = do
  rows <- Query (Int64, Int64) -> m [(Int64, Int64)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> [Int64] -> Query (Int64, Int64)
Tmpl.countDLQChildrenBatchSQL Text
schemaName Text
tableName [Int64]
ids)
  pure $ Map.fromList rows

-- ---------------------------------------------------------------------------
-- Job Dependency Operations
-- ---------------------------------------------------------------------------

-- | Suspend a parent's claimable children. In-flight ones are left alone. Returns the
-- number suspended.
pauseChildren
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> m Int64
pauseChildren :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
pauseChildren Text
schemaName Text
tableName Int64
parentJobId =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Query ()
Tmpl.pauseChildrenSQL Text
schemaName Text
tableName Int64
parentJobId)

-- | Resume a parent's suspended children. Returns the number resumed.
resumeChildren
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> m Int64
resumeChildren :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
resumeChildren Text
schemaName Text
tableName Int64
parentJobId =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Query ()
Tmpl.resumeChildrenSQL Text
schemaName Text
tableName Int64
parentJobId)

-- | Delete a job and every descendant under it, resuming the parent of a root that is
-- itself a child. Returns the number deleted.
cancelJobCascade
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Root job id
  -> m Int64
cancelJobCascade :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
cancelJobCascade = (Text -> Text -> Int64 -> Query Int64)
-> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
(Text -> Text -> Int64 -> Query Int64)
-> Text -> Text -> Int64 -> m Int64
cascadeDeleteJob Text -> Text -> Int64 -> Query Int64
Tmpl.cancelJobCascadeSQL

-- | Transactional wrapper for cascade-delete SQL. Reads the root's parent, runs the
-- supplied delete template, and wakes the parent for a completion round when
-- anything was deleted. 'cancelJobCascade' and 'forceCancelJob' share this shell.
cascadeDeleteJob
  :: (MonadArbiter m)
  => (SchemaName -> TableName -> Int64 -> Q.Query Int64)
  -- ^ Cascade-delete query builder (returns the deleted count).
  -> SchemaName
  -> TableName
  -> Int64
  -> m Int64
cascadeDeleteJob :: forall (m :: * -> *).
MonadArbiter m =>
(Text -> Text -> Int64 -> Query Int64)
-> Text -> Text -> Int64 -> m Int64
cascadeDeleteJob Text -> Text -> Int64 -> Query Int64
mkSql Text
schemaName Text
tableName Int64
jobId = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
  rootParentId <- Text -> Text -> Int64 -> m (Maybe Int64)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe Int64)
lockParentOf Text
schemaName Text
tableName Int64
jobId
  deleted <- countOr0 (mkSql schemaName tableName jobId)

  when (deleted > 0)
    $ for_ rootParentId
    $ tryResumeParent LocksHeld schemaName tableName

  pure deleted

-- | Delete a whole job tree, named by any node in it. Walks up to the root, then deletes
-- from there down. The root has no parent to resume. Returns the number deleted.
cancelJobTree
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Any job id in the tree
  -> m Int64
cancelJobTree :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
cancelJobTree Text
schemaName Text
tableName Int64
jobId =
  Text -> Query Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Query Int64 -> m Int64
countStrict Text
"cancelJobTree" (Text -> Text -> Int64 -> Query Int64
Tmpl.cancelJobTreeSQL Text
schemaName Text
tableName Int64
jobId)

-- | Cascade-cancel a job subtree. Flags still-live claimed jobs, deletes the rest,
-- and NOTIFYs the queue's cancel channel for every claimed job affected. Workers
-- async-cancel the matching handler thread on receipt.
forceCancelJob
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Root job id
  -> m Int64
forceCancelJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
forceCancelJob = (Text -> Text -> Int64 -> Query Int64)
-> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
(Text -> Text -> Int64 -> Query Int64)
-> Text -> Text -> Int64 -> m Int64
cascadeDeleteJob Text -> Text -> Int64 -> Query Int64
Tmpl.forceCancelJobSQL

-- ---------------------------------------------------------------------------
-- Suspend/Resume Operations
-- ---------------------------------------------------------------------------

-- | Suspend a job, making it unclaimable. Refuses an in-flight job.
suspendJob
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Job id
  -> m Int64
suspendJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
suspendJob Text
schemaName Text
tableName Int64
jobId =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Query ()
Tmpl.suspendJobSQL Text
schemaName Text
tableName Int64
jobId)

-- | Resume a suspended job, making it claimable again.
resumeJob
  :: (MonadArbiter m)
  => Text
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Job id
  -> m Int64
resumeJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
resumeJob Text
schemaName Text
tableName Int64
jobId =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Int64 -> Query ()
Tmpl.resumeJobSQL Text
schemaName Text
tableName Int64
jobId)

-- | Where a queue's next groups pass resumes. The summary window and the emptied scan
-- walk the key space independently.
data GroupsCursor = GroupsCursor
  { GroupsCursor -> Maybe Text
groupsWindowFrom :: Maybe Text
  , GroupsCursor -> Maybe Text
groupsEmptiedFrom :: Maybe Text
  }
  deriving stock (GroupsCursor -> GroupsCursor -> Bool
(GroupsCursor -> GroupsCursor -> Bool)
-> (GroupsCursor -> GroupsCursor -> Bool) -> Eq GroupsCursor
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: GroupsCursor -> GroupsCursor -> Bool
== :: GroupsCursor -> GroupsCursor -> Bool
$c/= :: GroupsCursor -> GroupsCursor -> Bool
/= :: GroupsCursor -> GroupsCursor -> Bool
Eq, Int -> GroupsCursor -> ShowS
[GroupsCursor] -> ShowS
GroupsCursor -> String
(Int -> GroupsCursor -> ShowS)
-> (GroupsCursor -> String)
-> ([GroupsCursor] -> ShowS)
-> Show GroupsCursor
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> GroupsCursor -> ShowS
showsPrec :: Int -> GroupsCursor -> ShowS
$cshow :: GroupsCursor -> String
show :: GroupsCursor -> String
$cshowList :: [GroupsCursor] -> ShowS
showList :: [GroupsCursor] -> ShowS
Show)

instance ToJSON GroupsCursor where
  toJSON :: GroupsCursor -> Value
toJSON (GroupsCursor Maybe Text
window Maybe Text
emptied) =
    [Pair] -> Value
object ([Maybe Pair] -> [Pair]
forall a. [Maybe a] -> [a]
catMaybes [(Key
"window" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.=) (Text -> Pair) -> Maybe Text -> Maybe Pair
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text
window, (Key
"emptied" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.=) (Text -> Pair) -> Maybe Text -> Maybe Pair
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text
emptied])

instance FromJSON GroupsCursor where
  parseJSON :: Value -> Parser GroupsCursor
parseJSON = String
-> (Object -> Parser GroupsCursor) -> Value -> Parser GroupsCursor
forall a. String -> (Object -> Parser a) -> Value -> Parser a
withObject String
"GroupsCursor" ((Object -> Parser GroupsCursor) -> Value -> Parser GroupsCursor)
-> (Object -> Parser GroupsCursor) -> Value -> Parser GroupsCursor
forall a b. (a -> b) -> a -> b
$ \Object
obj ->
    Maybe Text -> Maybe Text -> GroupsCursor
GroupsCursor (Maybe Text -> Maybe Text -> GroupsCursor)
-> Parser (Maybe Text) -> Parser (Maybe Text -> GroupsCursor)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Object
obj Object -> Key -> Parser (Maybe Text)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"window" Parser (Maybe Text -> GroupsCursor)
-> Parser (Maybe Text) -> Parser GroupsCursor
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Object
obj Object -> Key -> Parser (Maybe Text)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"emptied"

-- | What one queue's groups pass did: the rows it rewrote, whether its missing-summary
-- repair threw, and where the next pass resumes.
data GroupsPass = GroupsPass
  { GroupsPass -> Int64
passRewritten :: Int64
  , GroupsPass -> Bool
passRepairFailed :: Bool
  , GroupsPass -> Maybe GroupsCursor
passResume :: Maybe GroupsCursor
  }

-- | 'Nothing' once both walks have wrapped. The queue starts over at its first key.
resumeCursor :: Maybe Text -> Maybe Text -> Maybe GroupsCursor
resumeCursor :: Maybe Text -> Maybe Text -> Maybe GroupsCursor
resumeCursor Maybe Text
Nothing Maybe Text
Nothing = Maybe GroupsCursor
forall a. Maybe a
Nothing
resumeCursor Maybe Text
window Maybe Text
emptied = GroupsCursor -> Maybe GroupsCursor
forall a. a -> Maybe a
Just (Maybe Text -> Maybe Text -> GroupsCursor
GroupsCursor Maybe Text
window Maybe Text
emptied)

-- | Recompute the groups table from the main queue, over one bounded batch of rows past
-- @cursor@. Locks the window's groups rows and the emptied ones (FOR UPDATE SKIP LOCKED),
-- then rewrites them, which deletes the emptied. The missing-summary repair runs in its
-- own transaction. The caller owns any cross-pool coordination (see 'runGatedState'
-- and 'refreshAllGroups').
refreshGroupsForQueue
  :: (MonadArbiter m)
  => SchemaName
  -> TableName
  -> Int
  -- ^ Rows this pass covers.
  -> Maybe GroupsCursor
  -- ^ Resume past these keys, or start at the first.
  -> m GroupsPass
refreshGroupsForQueue :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int -> Maybe GroupsCursor -> m GroupsPass
refreshGroupsForQueue Text
schemaName Text
tableName Int
batch Maybe GroupsCursor
resume = do
  (rewritten, windowEnd, nextEmptied) <- m (Int64, Maybe Text, Maybe Text)
-> m (Int64, Maybe Text, Maybe Text)
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Int64, Maybe Text, Maybe Text)
 -> m (Int64, Maybe Text, Maybe Text))
-> m (Int64, Maybe Text, Maybe Text)
-> m (Int64, Maybe Text, Maybe Text)
forall a b. (a -> b) -> a -> b
$ do
    window <- Query Text -> m [Text]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int -> Maybe Text -> Query Text
Tmpl.groupsWindowSQL Text
schemaName Text
tableName Int
batch Maybe Text
cursor)
    emptied <- MA.executeQuery (Tmpl.emptiedWindowSQL schemaName tableName batch emptiedCursor)
    let upper = [Text] -> Maybe Text
forall a. [a] -> Maybe a
lastRow [Text]
window
    locked <- MA.executeQuery (Tmpl.lockGroupsSQL schemaName tableName cursor upper emptied)
    rewrittenCount <-
      if null locked
        then pure 0
        else sum <$> MA.executeQuery (Tmpl.refreshGroupsSQL schemaName tableName locked)
    -- A short emptied batch reached the last emptied key.
    pure (rewrittenCount, upper, if length emptied < batch then Nothing else lastRow emptied)
  repair <-
    tryAny . withDbTransaction $
      MA.executeQuery (Tmpl.insertMissingGroupsSQL schemaName tableName batch cursor windowEnd)
  let repaired = [(Text, Bool)]
-> Either SomeException [(Text, Bool)] -> [(Text, Bool)]
forall b a. b -> Either a b -> b
fromRight [] Either SomeException [(Text, Bool)]
repair
      -- A filled repair batch left keys behind.
      nextWindow = if [(Text, Bool)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [(Text, Bool)]
repaired Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
batch then [Text] -> Maybe Text
forall a. [a] -> Maybe a
lastRow (((Text, Bool) -> Text) -> [(Text, Bool)] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (Text, Bool) -> Text
forall a b. (a, b) -> a
fst [(Text, Bool)]
repaired) else Maybe Text
windowEnd
  pure
    GroupsPass
      { passRewritten = rewritten + fromIntegral (length (filter snd repaired))
      , passRepairFailed = isLeft repair
      , passResume = resumeCursor nextWindow nextEmptied
      }
  where
    cursor :: Maybe Text
cursor = Maybe GroupsCursor
resume Maybe GroupsCursor -> (GroupsCursor -> Maybe Text) -> Maybe Text
forall a b. Maybe a -> (a -> Maybe b) -> Maybe b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= GroupsCursor -> Maybe Text
groupsWindowFrom
    emptiedCursor :: Maybe Text
emptiedCursor = Maybe GroupsCursor
resume Maybe GroupsCursor -> (GroupsCursor -> Maybe Text) -> Maybe Text
forall a b. Maybe a -> (a -> Maybe b) -> Maybe b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= GroupsCursor -> Maybe Text
groupsEmptiedFrom

-- | The last row, in the order the database returned it.
lastRow :: [a] -> Maybe a
lastRow :: forall a. [a] -> Maybe a
lastRow = (Maybe a -> a -> Maybe a) -> Maybe a -> [a] -> Maybe a
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' (\Maybe a
_ a
row -> a -> Maybe a
forall a. a -> Maybe a
Just a
row) Maybe a
forall a. Maybe a
Nothing

-- | Groups rows one refresh transaction recomputes, split across the queues sharing
-- it.
groupsRefreshBatch :: Int
groupsRefreshBatch :: Int
groupsRefreshBatch = Int
20000

-- | Schema-wide groups-table refresh, 'groupsRefreshBatch' rows for the pass. Wrap in
-- 'runGatedState' so one pool runs it per interval and every pool resumes from the
-- same cursors. A caller that discards the cursors refreshes the same head of each
-- table forever. Each queue runs in a savepoint. One queue's failure leaves the rest.
-- Its row locks stand until the caller's transaction ends. Returns the rows rewritten,
-- the queue names that failed or whose repair failed, and where each queue resumes.
refreshAllGroups
  :: (MonadArbiter m)
  => SchemaName
  -> [TableName]
  -> Map.Map TableName GroupsCursor
  -- ^ Where the previous pass left off, per queue.
  -> m ((Int64, [Text]), Map.Map TableName GroupsCursor)
refreshAllGroups :: forall (m :: * -> *).
MonadArbiter m =>
Text
-> [Text]
-> Map Text GroupsCursor
-> m ((Int64, [Text]), Map Text GroupsCursor)
refreshAllGroups Text
schemaName [Text]
queues Map Text GroupsCursor
cursors = do
  (refreshed, failed) <- (Text -> Text -> m GroupsPass)
-> Text -> [Text] -> m ([(Text, GroupsPass)], [Text])
forall (m :: * -> *) a.
MonadUnliftIO m =>
(Text -> Text -> m a) -> Text -> [Text] -> m ([(Text, a)], [Text])
sweepEachQueue Text -> Text -> m GroupsPass
one Text
schemaName [Text]
queues
  let rewritten = [Int64] -> Int64
forall a. Num a => [a] -> a
forall (t :: * -> *) a. (Foldable t, Num a) => t a -> a
sum [GroupsPass -> Int64
passRewritten GroupsPass
pass | (Text
_, GroupsPass
pass) <- [(Text, GroupsPass)]
refreshed]
      unrepaired = [Text
tbl | (Text
tbl, GroupsPass
pass) <- [(Text, GroupsPass)]
refreshed, GroupsPass -> Bool
passRepairFailed GroupsPass
pass]
      resumed = (Map Text GroupsCursor
 -> (Text, GroupsPass) -> Map Text GroupsCursor)
-> Map Text GroupsCursor
-> [(Text, GroupsPass)]
-> Map Text GroupsCursor
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' (\Map Text GroupsCursor
acc (Text
tbl, GroupsPass
pass) -> (Maybe GroupsCursor -> Maybe GroupsCursor)
-> Text -> Map Text GroupsCursor -> Map Text GroupsCursor
forall k a.
Ord k =>
(Maybe a -> Maybe a) -> k -> Map k a -> Map k a
Map.alter (Maybe GroupsCursor -> Maybe GroupsCursor -> Maybe GroupsCursor
forall a b. a -> b -> a
const (GroupsPass -> Maybe GroupsCursor
passResume GroupsPass
pass)) Text
tbl Map Text GroupsCursor
acc) Map Text GroupsCursor
cursors [(Text, GroupsPass)]
refreshed
  pure ((rewritten, failed <> unrepaired), resumed)
  where
    perQueue :: Int
perQueue = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
1 (Int
groupsRefreshBatch Int -> Int -> Int
forall a. Integral a => a -> a -> a
`div` Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
1 ([Text] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Text]
queues))
    one :: Text -> Text -> m GroupsPass
one Text
schema Text
tbl = Text -> Text -> Int -> Maybe GroupsCursor -> m GroupsPass
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int -> Maybe GroupsCursor -> m GroupsPass
refreshGroupsForQueue Text
schema Text
tbl Int
perQueue (Text -> Map Text GroupsCursor -> Maybe GroupsCursor
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Text
tbl Map Text GroupsCursor
cursors)

-- | Run 'refreshAllGroups' until it scans each queue's complete groups table.
-- Use one batch and transaction for each pass. This is a repair operation. The
-- reaper runs one batch at each tick.
refreshAllGroupsFully
  :: (MonadArbiter m)
  => SchemaName
  -> [TableName]
  -> m (Int64, [Text])
refreshAllGroupsFully :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
refreshAllGroupsFully Text
schemaName [Text]
queues = do
  (walked, failed) <- (Text -> Text -> m (Int64, Bool))
-> Text -> [Text] -> m ([(Text, (Int64, Bool))], [Text])
forall (m :: * -> *) a.
MonadUnliftIO m =>
(Text -> Text -> m a) -> Text -> [Text] -> m ([(Text, a)], [Text])
sweepEachQueue Text -> Text -> m (Int64, Bool)
forall {m :: * -> *}.
MonadArbiter m =>
Text -> Text -> m (Int64, Bool)
walk Text
schemaName [Text]
queues
  let unrepaired = [Text
tbl | (Text
tbl, (Int64
_, Bool
True)) <- [(Text, (Int64, Bool))]
walked]
  pure (sum [count | (_, (count, _)) <- walked], failed <> unrepaired)
  where
    walk :: Text -> Text -> m (Int64, Bool)
walk Text
schema Text
tbl = Int64 -> Bool -> Maybe GroupsCursor -> m (Int64, Bool)
go Int64
0 Bool
False Maybe GroupsCursor
forall a. Maybe a
Nothing
      where
        go :: Int64 -> Bool -> Maybe GroupsCursor -> m (Int64, Bool)
go Int64
acc Bool
failedRepair Maybe GroupsCursor
cursor = do
          pass <- Text -> Text -> Int -> Maybe GroupsCursor -> m GroupsPass
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int -> Maybe GroupsCursor -> m GroupsPass
refreshGroupsForQueue Text
schema Text
tbl Int
groupsRefreshBatch Maybe GroupsCursor
cursor
          let total = Int64
acc Int64 -> Int64 -> Int64
forall a. Num a => a -> a -> a
+ GroupsPass -> Int64
passRewritten GroupsPass
pass
              failedYet = Bool
failedRepair Bool -> Bool -> Bool
|| GroupsPass -> Bool
passRepairFailed GroupsPass
pass
          maybe (pure (total, failedYet)) (go total failedYet . Just) (passResume pass)

-- | Run a per-queue sweep over every queue, returning what each one swept and the
-- names of queues whose sweep threw.
sweepEachQueue
  :: (MonadUnliftIO m)
  => (SchemaName -> TableName -> m a)
  -> SchemaName
  -> [TableName]
  -> m ([(TableName, a)], [Text])
sweepEachQueue :: forall (m :: * -> *) a.
MonadUnliftIO m =>
(Text -> Text -> m a) -> Text -> [Text] -> m ([(Text, a)], [Text])
sweepEachQueue Text -> Text -> m a
sweepOne Text
schemaName [Text]
queues = do
  (failures, swept) <- [Either Text (Text, a)] -> ([Text], [(Text, a)])
forall a b. [Either a b] -> ([a], [b])
partitionEithers ([Either Text (Text, a)] -> ([Text], [(Text, a)]))
-> m [Either Text (Text, a)] -> m ([Text], [(Text, a)])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (Text -> m (Either Text (Text, a)))
-> [Text] -> m [Either Text (Text, a)]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse Text -> m (Either Text (Text, a))
run [Text]
queues
  pure (swept, failures)
  where
    run :: Text -> m (Either Text (Text, a))
run Text
queue = (SomeException -> Text)
-> (a -> (Text, a))
-> Either SomeException a
-> Either Text (Text, a)
forall a b c d. (a -> b) -> (c -> d) -> Either a c -> Either b d
forall (p :: * -> * -> *) a b c d.
Bifunctor p =>
(a -> b) -> (c -> d) -> p a c -> p b d
bimap (Text -> SomeException -> Text
forall a b. a -> b -> a
const Text
queue) (Text
queue,) (Either SomeException a -> Either Text (Text, a))
-> m (Either SomeException a) -> m (Either Text (Text, a))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m a -> m (Either SomeException a)
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> m (Either SomeException a)
tryAny (Text -> Text -> m a
sweepOne Text
schemaName Text
queue)

-- | 'sweepEachQueue' for a sweep reporting the rows it touched, totalled.
sweepQueues
  :: (MonadUnliftIO m)
  => (SchemaName -> TableName -> m Int64)
  -> SchemaName
  -> [TableName]
  -> m (Int64, [Text])
sweepQueues :: forall (m :: * -> *).
MonadUnliftIO m =>
(Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
sweepQueues Text -> Text -> m Int64
sweepOne Text
schemaName [Text]
queues =
  ([(Text, Int64)] -> Int64)
-> ([(Text, Int64)], [Text]) -> (Int64, [Text])
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first ([Int64] -> Int64
forall a. Num a => [a] -> a
forall (t :: * -> *) a. (Foldable t, Num a) => t a -> a
sum ([Int64] -> Int64)
-> ([(Text, Int64)] -> [Int64]) -> [(Text, Int64)] -> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ((Text, Int64) -> Int64) -> [(Text, Int64)] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map (Text, Int64) -> Int64
forall a b. (a, b) -> b
snd) (([(Text, Int64)], [Text]) -> (Int64, [Text]))
-> m ([(Text, Int64)], [Text]) -> m (Int64, [Text])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (Text -> Text -> m Int64)
-> Text -> [Text] -> m ([(Text, Int64)], [Text])
forall (m :: * -> *) a.
MonadUnliftIO m =>
(Text -> Text -> m a) -> Text -> [Text] -> m ([(Text, a)], [Text])
sweepEachQueue Text -> Text -> m Int64
sweepOne Text
schemaName [Text]
queues

-- | Sweep exhausted jobs across all queues. Returns the total moved and the
-- names of queues whose sweep failed.
sweepExhaustedJobs
  :: (MonadArbiter m)
  => SchemaName
  -> [TableName]
  -> m (Int64, [Text])
sweepExhaustedJobs :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
sweepExhaustedJobs = (Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadUnliftIO m =>
(Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
sweepQueues Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
sweepExhaustedForQueue

-- | Move each exhausted job to the DLQ via the tree-aware 'moveToDLQFields'. One
-- transaction for the pass, taking every parent and every tree the moves will touch
-- up front.
sweepExhaustedForQueue
  :: (MonadArbiter m)
  => SchemaName
  -> TableName
  -> m Int64
sweepExhaustedForQueue :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
sweepExhaustedForQueue Text
schemaName Text
tableName = m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
  exhausted <- Query (Int64, Int64, Maybe Int64, Bool)
-> m [(Int64, Int64, Maybe Int64, Bool)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int -> Query (Int64, Int64, Maybe Int64, Bool)
Tmpl.selectExhaustedJobsSQL Text
schemaName Text
tableName Int
exhaustedSweepBatch)
  let ids = [Int64
jobId | (Int64
jobId, Int64
_, Maybe Int64
_, Bool
_) <- [(Int64, Int64, Maybe Int64, Bool)]
exhausted]
  lockJobParents schemaName tableName [mParentId | (_, _, mParentId, _) <- exhausted]
  lockJobTrees schemaName tableName ids
  getSum <$> getAp (foldMap moveOne exhausted)
  where
    moveOne :: (Int64, Int64, Maybe Int64, Bool) -> Ap m (Sum Int64)
moveOne (Int64
jobId, Int64
cseq, Maybe Int64
mParentId, Bool
rollup) =
      m (Sum Int64) -> Ap m (Sum Int64)
forall {k} (f :: k -> *) (a :: k). f a -> Ap f a
Ap (m (Sum Int64) -> Ap m (Sum Int64))
-> m (Sum Int64) -> Ap m (Sum Int64)
forall a b. (a -> b) -> a -> b
$
        Int64 -> Sum Int64
forall a. a -> Sum a
Sum (Int64 -> Sum Int64)
-> (Either SomeException Int64 -> Int64)
-> Either SomeException Int64
-> Sum Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int64 -> Either SomeException Int64 -> Int64
forall b a. b -> Either a b -> b
fromRight Int64
0
          (Either SomeException Int64 -> Sum Int64)
-> m (Either SomeException Int64) -> m (Sum Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m Int64 -> m (Either SomeException Int64)
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> m (Either SomeException a)
tryAny (TreeLocks
-> DLQMove
-> Text
-> Text
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> m Int64
forall (m :: * -> *).
MonadArbiter m =>
TreeLocks
-> DLQMove
-> Text
-> Text
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> m Int64
moveToDLQFields TreeLocks
LocksHeld DLQMove
Tmpl.MoveIfExhausted Text
schemaName Text
tableName Text
sweepError Int64
jobId Int64
cseq Maybe Int64
mParentId Bool
rollup)
    sweepError :: Text
sweepError = Text
"max attempts exceeded (reaper sweep)"

-- | Per-queue cap on jobs swept to the DLQ in one reaper pass.
exhaustedSweepBatch :: Int
exhaustedSweepBatch :: Int
exhaustedSweepBatch = Int
1000

-- | Sweep force-cancel-flagged jobs whose lease has lapsed across all queues.
-- A live worker's jobs are left for the worker's own cancel handler. Returns the
-- total deleted and the names of queues whose sweep failed.
sweepCancelledJobs
  :: (MonadArbiter m)
  => SchemaName
  -> [TableName]
  -> m (Int64, [Text])
sweepCancelledJobs :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
sweepCancelledJobs = (Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadUnliftIO m =>
(Text -> Text -> m Int64) -> Text -> [Text] -> m (Int64, [Text])
sweepQueues Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
sweepCancelledForQueue

-- | Delete one queue's lease-lapsed flagged jobs, resuming any parents left
-- childless. 'deleteCancelledJobs' runs in its own transaction and re-checks
-- the lease under the row lock.
sweepCancelledForQueue
  :: (MonadArbiter m)
  => SchemaName
  -> TableName
  -> m Int64
sweepCancelledForQueue :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
sweepCancelledForQueue Text
schemaName Text
tableName = do
  ids <- Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int -> Query Int64
Tmpl.selectCancelledReapableJobsSQL Text
schemaName Text
tableName Int
cancelledSweepBatch)
  fromIntegral . length <$> deleteCancelledJobs schemaName tableName Nothing ids

-- | Per-queue cap on flagged jobs reaped in one pass.
cancelledSweepBatch :: Int
cancelledSweepBatch :: Int
cancelledSweepBatch = Int
1000

-- | Upsert a cron schedule's default expression and overlap policy, preserving user
-- overrides and the enabled flag. @queue_name@ is overwritten on conflict.
upsertCronDefault
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Text
  -- ^ Schedule name
  -> Text
  -- ^ Queue name
  -> Text
  -- ^ Default cron expression
  -> Text
  -- ^ Default overlap policy
  -> Maybe Text
  -- ^ Default IANA tz name (@Nothing@ = UTC).
  -> m Int64
upsertCronDefault :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> Text -> Text -> Maybe Text -> m Int64
upsertCronDefault Text
schemaName Text
scheduleName Text
queueName Text
defaultExpr Text
defaultOv Maybe Text
defaultTz =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Text -> Text -> Text -> Maybe Text -> Query ()
Tmpl.upsertCronDefaultSQL Text
schemaName Text
scheduleName Text
queueName Text
defaultExpr Text
defaultOv Maybe Text
defaultTz)

-- | List cron schedules ordered by name, optionally filtered by queue.
listCronSchedules
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Maybe Text
  -- ^ Queue filter. 'Nothing' returns schedules for all queues.
  -> m [CronScheduleRow]
listCronSchedules :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> m [CronScheduleRow]
listCronSchedules Text
schemaName Maybe Text
mQueue =
  Query CronScheduleRow -> m [CronScheduleRow]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Maybe Text -> Query CronScheduleRow
Tmpl.listCronSchedulesSQL Text
schemaName Maybe Text
mQueue)

-- | Get a single cron schedule by name.
getCronScheduleByName
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Text
  -- ^ Schedule name
  -> m (Maybe CronScheduleRow)
getCronScheduleByName :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
getCronScheduleByName Text
schemaName Text
scheduleName = do
  rows <- Query CronScheduleRow -> m [CronScheduleRow]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query CronScheduleRow
Tmpl.getCronScheduleByNameSQL Text
schemaName Text
scheduleName)
  pure (listToMaybe rows)

-- | Patch a cron schedule. Returns rows affected, 0 for a name that is not there.
updateCronSchedule
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Text
  -- ^ Schedule name
  -> CronScheduleUpdate
  -> m Int64
updateCronSchedule :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> CronScheduleUpdate -> m Int64
updateCronSchedule Text
schemaName Text
scheduleName CronScheduleUpdate
upd =
  m Int64 -> (Query () -> m Int64) -> Maybe (Query ()) -> m Int64
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0) Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Text -> CronScheduleUpdate -> Maybe (Query ())
Tmpl.updateCronScheduleSQL Text
schemaName Text
scheduleName CronScheduleUpdate
upd)

-- | Update @last_fired_at@ to NOW() for a cron schedule.
touchCronLastFired
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Text
  -- ^ Schedule name
  -> m Int64
touchCronLastFired :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
touchCronLastFired Text
schemaName Text
scheduleName =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Query ()
Tmpl.touchCronLastFiredSQL Text
schemaName Text
scheduleName)

-- | Advance @last_checked_at@ to the supplied watermark for the given cron
-- schedule names. The watermark is the minute boundary the scheduler finished
-- evaluating. A wrapping @GREATEST@ in the SQL keeps the column monotonic when
-- concurrent worker pools race.
touchCronChecked
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> UTCTime
  -- ^ Watermark (the minute the scheduler is advancing to)
  -> [Text]
  -- ^ Schedule names
  -> m Int64
touchCronChecked :: forall (m :: * -> *).
MonadArbiter m =>
Text -> UTCTime -> [Text] -> m Int64
touchCronChecked Text
_ UTCTime
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
touchCronChecked Text
schemaName UTCTime
watermark [Text]
names =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> UTCTime -> [Text] -> Query ()
Tmpl.touchCronCheckedSQL Text
schemaName UTCTime
watermark [Text]
names)

-- | Claim a minute floor for a schedule. 'True' when the caller proceeds with the
-- insert. 'False' when another pool fired this minute.
tryFireCronGate
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Text
  -- ^ Schedule name
  -> UTCTime
  -- ^ Minute floor for the tick being attempted
  -> m Bool
tryFireCronGate :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> UTCTime -> m Bool
tryFireCronGate Text
schemaName Text
scheduleName UTCTime
minuteFloor = do
  rows <- Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> UTCTime -> Text -> Query ()
Tmpl.tryFireCronGateSQL Text
schemaName UTCTime
minuteFloor Text
scheduleName)
  pure (rows > 0)

-- | Try to acquire the (schema, queue, name) cron leader lock. Must be inside a transaction.
tryAcquireCronLeader
  :: (MonadArbiter m)
  => SchemaName
  -> Text
  -- ^ Queue name
  -> Text
  -- ^ Schedule name
  -> m Bool
tryAcquireCronLeader :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> m Bool
tryAcquireCronLeader Text
schemaName Text
queueName Text
scheduleName = do
  rows <- Query Bool -> m [Bool]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Text -> Query Bool
Tmpl.tryAcquireCronLeaderSQL Text
schemaName Text
queueName Text
scheduleName)
  pure (fromMaybe False (listToMaybe rows))

-- | Result of a manual run request.
data RunRequestOutcome = RunReqNotFound | RunReqDisabled | RunReqStamped | RunReqPending
  deriving stock (RunRequestOutcome -> RunRequestOutcome -> Bool
(RunRequestOutcome -> RunRequestOutcome -> Bool)
-> (RunRequestOutcome -> RunRequestOutcome -> Bool)
-> Eq RunRequestOutcome
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: RunRequestOutcome -> RunRequestOutcome -> Bool
== :: RunRequestOutcome -> RunRequestOutcome -> Bool
$c/= :: RunRequestOutcome -> RunRequestOutcome -> Bool
/= :: RunRequestOutcome -> RunRequestOutcome -> Bool
Eq, Int -> RunRequestOutcome -> ShowS
[RunRequestOutcome] -> ShowS
RunRequestOutcome -> String
(Int -> RunRequestOutcome -> ShowS)
-> (RunRequestOutcome -> String)
-> ([RunRequestOutcome] -> ShowS)
-> Show RunRequestOutcome
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> RunRequestOutcome -> ShowS
showsPrec :: Int -> RunRequestOutcome -> ShowS
$cshow :: RunRequestOutcome -> String
show :: RunRequestOutcome -> String
$cshowList :: [RunRequestOutcome] -> ShowS
showList :: [RunRequestOutcome] -> ShowS
Show)

-- | Stamp a manual run request on an enabled schedule and NOTIFY the run-now
-- channel. A pending, unexpired request is left as it stands.
requestCronRun
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Text
  -- ^ Schedule name
  -> m RunRequestOutcome
requestCronRun :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
requestCronRun Text
schemaName Text
scheduleName = do
  rows <- Query Text -> m [Text]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query Text
Tmpl.requestCronRunSQL Text
schemaName Text
scheduleName)
  pure $ case listToMaybe rows of
    Just Text
"stamped" -> RunRequestOutcome
RunReqStamped
    Just Text
"pending" -> RunRequestOutcome
RunReqPending
    Just Text
"disabled" -> RunRequestOutcome
RunReqDisabled
    Maybe Text
_ -> RunRequestOutcome
RunReqNotFound

-- | Claim a pending run request, returning the claimed row. 'Nothing' when another
-- pool won the claim or the schedule is disabled.
claimCronRun
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> Text
  -- ^ Schedule name
  -> m (Maybe CronScheduleRow)
claimCronRun :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
claimCronRun Text
schemaName Text
scheduleName = do
  rows <- Query CronScheduleRow -> m [CronScheduleRow]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query CronScheduleRow
Tmpl.claimCronRunSQL Text
schemaName Text
scheduleName)
  pure $ listToMaybe rows

-- | Record when a manual run last fired a job for a cron schedule.
touchCronManualRun
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> UTCTime
  -- ^ When the manual run fired
  -> Text
  -- ^ Schedule name
  -> m Int64
touchCronManualRun :: forall (m :: * -> *).
MonadArbiter m =>
Text -> UTCTime -> Text -> m Int64
touchCronManualRun Text
schemaName UTCTime
firedAt Text
scheduleName =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> UTCTime -> Text -> Query ()
Tmpl.touchCronManualRunSQL Text
schemaName UTCTime
firedAt Text
scheduleName)

-- | Enabled schedules among @names@ that have a pending run request.
pendingCronRuns
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> [Text]
  -- ^ Schedule names
  -> m [Text]
pendingCronRuns :: forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m [Text]
pendingCronRuns Text
_ [] = [Text] -> m [Text]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
pendingCronRuns Text
schemaName [Text]
names =
  Query Text -> m [Text]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> [Text] -> Query Text
Tmpl.pendingCronRunsSQL Text
schemaName [Text]
names)

-- ---------------------------------------------------------------------------
-- Queue Operations
-- ---------------------------------------------------------------------------

-- | Insert an arbiter_queues row with defaults when absent.
ensureQueue
  :: (MonadArbiter m)
  => SchemaName
  -> Text
  -- ^ Queue name
  -> m Int64
ensureQueue :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
ensureQueue Text
schemaName Text
queue =
  Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
    (Text -> Text -> Query ()
Tmpl.ensureQueueSQL Text
schemaName Text
queue)

-- | Set the queue's @paused@ flag, creating the row if missing.
setQueuePaused
  :: (MonadArbiter m)
  => SchemaName
  -> Text
  -- ^ Queue name
  -> Bool
  -> m Int64
setQueuePaused :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Bool -> m Int64
setQueuePaused Text
schemaName Text
queue Bool
paused =
  Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 (Text -> Text -> Bool -> Query Int64
Tmpl.setQueuePausedSQL Text
schemaName Text
queue Bool
paused)

-- | Get the arbiter_queues row for a single queue. 'Nothing' when absent.
getQueue
  :: (MonadArbiter m)
  => SchemaName
  -> Text
  -- ^ Queue name
  -> m (Maybe QueueRow)
getQueue :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe QueueRow)
getQueue Text
schemaName Text
queue = do
  rows <- Query QueueRow -> m [QueueRow]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query QueueRow
Tmpl.getQueueSQL Text
schemaName Text
queue)
  pure $ listToMaybe rows

-- | List all arbiter_queues rows, ordered by queue name.
listQueues
  :: (MonadArbiter m)
  => SchemaName
  -> m [QueueRow]
listQueues :: forall (m :: * -> *). MonadArbiter m => Text -> m [QueueRow]
listQueues Text
schemaName =
  Query QueueRow -> m [QueueRow]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Query QueueRow
Tmpl.listQueuesSQL Text
schemaName)

-- | Read child results, DLQ errors, and the parent_state snapshot for a rollup
-- finalizer in a single query.
readChildResultsRaw
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Parent job id
  -> m (Map.Map Int64 Value, Map.Map Int64 Text, Maybe Value, Map.Map Int64 Text)
readChildResultsRaw :: forall (m :: * -> *).
MonadArbiter m =>
Text
-> Text
-> Int64
-> m (Map Int64 Value, Map Int64 Text, Maybe Value, Map Int64 Text)
readChildResultsRaw Text
schemaName Text
tableName Int64
parentJobId = do
  rows <- Query (Text, Maybe Int64, Maybe Value, Maybe Text, Maybe Int64)
-> m [(Text, Maybe Int64, Maybe Value, Maybe Text, Maybe Int64)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text
-> Text
-> Int64
-> Query (Text, Maybe Int64, Maybe Value, Maybe Text, Maybe Int64)
Tmpl.readChildResultsSQL Text
schemaName Text
tableName Int64
parentJobId)
  foldM parseRow (Map.empty, Map.empty, Nothing, Map.empty) rows
  where
    parseRow :: (Map k a, Map k a, Maybe a, Map k a)
-> (a, Maybe k, Maybe a, Maybe a, Maybe k)
-> f (Map k a, Map k a, Maybe a, Map k a)
parseRow (!Map k a
results, !Map k a
errors, !Maybe a
snap, !Map k a
dlqFailures) (a, Maybe k, Maybe a, Maybe a, Maybe k)
row = case (a, Maybe k, Maybe a, Maybe a, Maybe k)
row of
      (a
"r", Just k
cid, Just a
val, Maybe a
_, Maybe k
_) ->
        (Map k a, Map k a, Maybe a, Map k a)
-> f (Map k a, Map k a, Maybe a, Map k a)
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (k -> a -> Map k a -> Map k a
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert k
cid a
val Map k a
results, Map k a
errors, Maybe a
snap, Map k a
dlqFailures)
      (a
"e", Just k
jid, Maybe a
_, Just a
err, Just k
dlqPk) ->
        (Map k a, Map k a, Maybe a, Map k a)
-> f (Map k a, Map k a, Maybe a, Map k a)
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Map k a
results, k -> a -> Map k a -> Map k a
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert k
jid a
err Map k a
errors, Maybe a
snap, k -> a -> Map k a -> Map k a
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert k
dlqPk a
err Map k a
dlqFailures)
      (a
"e", Just k
jid, Maybe a
_, Maybe a
Nothing, Just k
dlqPk) ->
        (Map k a, Map k a, Maybe a, Map k a)
-> f (Map k a, Map k a, Maybe a, Map k a)
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Map k a
results, k -> a -> Map k a -> Map k a
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert k
jid a
"" Map k a
errors, Maybe a
snap, k -> a -> Map k a -> Map k a
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert k
dlqPk a
"" Map k a
dlqFailures)
      (a
"s", Maybe k
_, Just a
val, Maybe a
_, Maybe k
_) ->
        (Map k a, Map k a, Maybe a, Map k a)
-> f (Map k a, Map k a, Maybe a, Map k a)
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Map k a
results, Map k a
errors, a -> Maybe a
forall a. a -> Maybe a
Just a
val, Map k a
dlqFailures)
      (a, Maybe k, Maybe a, Maybe a, Maybe k)
_ -> Text -> f (Map k a, Map k a, Maybe a, Map k a)
forall (m :: * -> *) a. MonadIO m => Text -> m a
throwParsing (Text -> f (Map k a, Map k a, Maybe a, Map k a))
-> Text -> f (Map k a, Map k a, Maybe a, Map k a)
forall a b. (a -> b) -> a -> b
$ Text
"readChildResultsRaw: unexpected row: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack ((a, Maybe k, Maybe a, Maybe a, Maybe k) -> String
forall a. Show a => a -> String
show (a, Maybe k, Maybe a, Maybe a, Maybe k)
row)

-- | Read a job's raw @parent_state@ snapshot, which a DLQ-retried finalizer comes back
-- carrying.
getParentStateSnapshot
  :: (MonadArbiter m)
  => SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> Int64
  -- ^ Job id
  -> m (Maybe Value)
getParentStateSnapshot :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe Value)
getParentStateSnapshot Text
schemaName Text
tableName Int64
jobId = do
  rows <- Query (Maybe Value) -> m [Maybe Value]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Query (Maybe Value)
Tmpl.getParentStateSnapshotSQL Text
schemaName Text
tableName Int64
jobId)
  pure (join (listToMaybe rows))

-- | Merge child results, DLQ errors and the snapshot, left-biased in that order.
mergeRawChildResults
  :: Map.Map Int64 Value
  -> Map.Map Int64 Text
  -> Maybe Value
  -> Map.Map Int64 (Either Text Value)
mergeRawChildResults :: Map Int64 Value
-> Map Int64 Text -> Maybe Value -> Map Int64 (Either Text Value)
mergeRawChildResults Map Int64 Value
results Map Int64 Text
failures Maybe Value
mSnapshot =
  (Text -> Either Text Value)
-> Map Int64 Text -> Map Int64 (Either Text Value)
forall a b k. (a -> b) -> Map k a -> Map k b
Map.map Text -> Either Text Value
forall a b. a -> Either a b
Left Map Int64 Text
failures
    Map Int64 (Either Text Value)
-> Map Int64 (Either Text Value) -> Map Int64 (Either Text Value)
forall k a. Ord k => Map k a -> Map k a -> Map k a
`Map.union` (Value -> Either Text Value)
-> Map Int64 Value -> Map Int64 (Either Text Value)
forall a b k. (a -> b) -> Map k a -> Map k b
Map.map Value -> Either Text Value
forall a b. b -> Either a b
Right Map Int64 Value
results
    Map Int64 (Either Text Value)
-> Map Int64 (Either Text Value) -> Map Int64 (Either Text Value)
forall k a. Ord k => Map k a -> Map k a -> Map k a
`Map.union` Map Int64 (Either Text Value)
base
  where
    base :: Map Int64 (Either Text Value)
base = case Maybe Value
mSnapshot of
      Just Value
val | Success Map Int64 (Either Text Value)
snapshotMap <- Value -> Result (Map Int64 (Either Text Value))
forall a. FromJSON a => Value -> Result a
fromJSON Value
val -> Map Int64 (Either Text Value)
snapshotMap
      Maybe Value
_ -> Map Int64 (Either Text Value)
forall k a. Map k a
Map.empty