{-# LANGUAGE AllowAmbiguousTypes #-}
{-# LANGUAGE OverloadedStrings #-}
{-# OPTIONS_HADDOCK not-home #-}
module Arbiter.Core.Operations
(
insertJob
, insertJobTreeNodeStamped
, insertJobTreeLeavesStamped
, spawnChildren
, insertJobsBatch
, insertJobsBatch_
, TraceStamp
, traceStamp
, insertResult
, insertResultsBatch
, getResultsByParent
, getDLQChildErrorsByParent
, persistParentState
, getParentStateSnapshot
, readChildResultsRaw
, mergeRawChildResults
, claimNextVisibleJobs
, claimNextVisibleJobsAs
, claimNextVisibleJobsBatched
, JobStatements
, mkJobStatements
, singleJobBatchSize
, noStagedCapacities
, claimJobsCached
, claimJobsBatchedCached
, RejectedRow
, deadLetterRejected
, decodeRow
, typedRow
, typedDLQRow
, typedArchiveRow
, addRateLimitTokens
, pruneRateLimitBuckets
, resetRateLimitBuckets
, wakeThrottledJobs
, wakeThrottledJobsForKey
, listRateLimitPolicies
, getRateLimitPolicy
, rateLimitPolicyExists
, listRateLimitBuckets
, updateRateLimitPolicyOverrides
, listConcurrencyPolicies
, getConcurrencyPolicy
, concurrencyPolicyExists
, listConcurrencyKeys
, updateConcurrencyPolicyOverrides
, pruneConcurrencyKeys
, reconcileConcurrencyCounts
, reconcileConcurrencyCountsIfStale
, reconcileAndPruneConcurrency
, ackJob
, statementsAck
, AckStatements
, ackJobWith
, ackJobsBatchWith
, ackJobsBatch
, archivesOnAck
, setVisibilityTimeout
, setVisibilityTimeoutBatch
, VisibilityUpdateInfo (..)
, updateJobForRetry
, nackJob
, nackJobsBatch
, lockJobParents
, lockJobRootsAndParents
, lockJobTrees
, lockJobTreesFromRoot
, TreesLocked
, TreeLocks (..)
, moveToDLQ
, moveToDLQFields
, moveToDLQBatch
, retryFromDLQ
, retryFromDLQWithPayload
, dlqJobExists
, listDLQJobs
, deleteDLQJob
, deleteDLQJobsBatch
, purgeDLQ
, retryFromDLQBatch
, retryDLQFiltered
, listArchiveJobs
, listArchiveFiltered
, getArchiveJobById
, listArchiveJobsByGroup
, countArchiveFiltered
, purgeArchives
, deleteArchiveJob
, deleteArchiveJobsBatch
, reEnqueueFromArchive
, reEnqueueFromArchiveWithPayload
, updateArchiveResult
, updateArchiveResultsBatch
, Tmpl.JobFilter (..)
, Tmpl.FilterTable (..)
, Tmpl.SortDir (..)
, Tmpl.sortDirName
, Tmpl.JobSortColumn (..)
, Tmpl.jobSortColumnName
, Tmpl.DLQSortColumn (..)
, Tmpl.dlqSortColumnName
, Tmpl.ArchiveSortColumn (..)
, Tmpl.archiveSortColumnName
, Tmpl.buildWhereClause
, listJobsFiltered
, listJobsFilteredOrdered
, listJobsWithStatus
, countJobsFiltered
, listDLQFiltered
, listDLQFilteredOrdered
, countDLQFiltered
, listDLQJobsByParent
, countDLQJobsByParent
, listJobs
, getJobById
, getJobByIdWithStatus
, getJobByDedupKey
, listJobsByGroup
, jobExists
, listJobsByParent
, countJobsByParent
, cancelJob
, cancelJobsBatch
, promoteJob
, rescheduleJob
, QueueStats (..)
, statsRowCodec
, queueStatusCounts
, QueueOverview (..)
, getQueueStats
, getAllQueueStats
, GroupSummary (..)
, listGroups
, countGroups
, countJobs
, countJobsByGroup
, countDLQJobs
, countChildrenBatch
, countDLQChildrenBatch
, pauseChildren
, resumeChildren
, cancelJobCascade
, cancelJobTree
, forceCancelJob
, CancelledScope (..)
, deleteCancelledJobs
, Tmpl.cancelTokenBump
, suspendJob
, resumeJob
, GroupsCursor (..)
, GroupsPass (..)
, refreshGroupsForQueue
, refreshAllGroups
, refreshAllGroupsFully
, sweepExhaustedJobs
, sweepCancelledJobs
, upsertCronDefault
, listCronSchedules
, getCronScheduleByName
, updateCronSchedule
, touchCronChecked
, tryFireCronGate
, tryAcquireCronLeader
, RunRequestOutcome (..)
, requestCronRun
, claimCronRun
, touchCronManualRun
, pendingCronRuns
, registerWorker
, heartbeatWorker
, setWorkerPaused
, markWorkerShuttingDown
, deregisterWorker
, workerRegistered
, listWorkers
, sweepStaleWorkers
, ensureQueue
, setQueuePaused
, setQueueDLQRetention
, getQueue
, listQueues
, schemaQueues
, runGated
, runGatedBounded
, runGatedShared
, runGatedState
, runGatedStateBounded
, setLocalStatementTimeout
, micros
, gateNameFor
, Shared (..)
) where
import Control.Monad (foldM, join, unless, void, when)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Data.Aeson
( FromJSON (..)
, ToJSON (..)
, Value
, object
, withObject
, (.:)
, (.:?)
, (.=)
)
import Data.Bifunctor (bimap, first)
import Data.Coerce (coerce)
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 (unsnoc)
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, isJust, listToMaybe, mapMaybe)
import Data.Sequence ((|>))
import Data.Sequence qualified as Seq
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Time (NominalDiffTime, UTCTime)
import Data.Tuple (swap)
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
, payloadWriteCodec
)
import Arbiter.Core.Concurrency.Spec
( ConcurrencyKey (..)
, HasConcurrency
, concurrencyFor
, concurrencyKeyText
)
import Arbiter.Core.Concurrency.Stats (ConcurrencyKeyView, ConcurrencyPolicyUpdate (..), ConcurrencyPolicyView)
import Arbiter.Core.CronSchedule (CronScheduleRow, CronScheduleUpdate (..))
import Arbiter.Core.Enum (enumFromText, showText)
import Arbiter.Core.Exceptions
( JobException (..)
, JobRetryableException (..)
, orThrowParsing
, throwJobGoneIds
, throwParsing
, throwScopedFailure
)
import Arbiter.Core.Job.Archive qualified as Archive
import Arbiter.Core.Job.DLQ qualified as DLQ
import Arbiter.Core.Job.Dedup (isReplaceDuplicate)
import Arbiter.Core.Job.Kind (HasKind, kindOf)
import Arbiter.Core.Job.Schema (SchemaName, TableName)
import Arbiter.Core.Job.Types
( ClaimSeq
, JobId
, JobPayload
, JobRead
, JobStatus (..)
, JobWrite
, PayloadColumns (..)
, Stored
, archiveFor
, attempts
, claimSeq
, claimedBy
, decodeStored
, dedupParts
, groupKey
, isRollup
, jobStatusFromText
, jobStatusToText
, mapPayload
, parentId
, payload
, primaryKey
)
import Arbiter.Core.Job.Types qualified as JT
import Arbiter.Core.Job.Types.Internal (JobRecord (..))
import Arbiter.Core.Json (decodeOrEmpty)
import Arbiter.Core.MonadArbiter (MonadArbiter, countOr0, countOr0Prepared, 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
)
import Arbiter.Core.RateLimit.Stats (RateLimitBucketView, RateLimitPolicyUpdate (..), RateLimitPolicyView)
import Arbiter.Core.Selector (runSelector, 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 (RowEdit, batchFrag, insertFrag, rowEdit)
import Arbiter.Core.Sql.Jobs qualified as Tmpl
import Arbiter.Core.Sql.Lifecycle qualified as Tmpl
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)
decodeRow :: (FromJSON payload) => JobRead (Stored payload) -> Either Text (JobRead payload)
decodeRow :: forall payload.
FromJSON payload =>
JobRead (Stored payload) -> Either Text (JobRead payload)
decodeRow JobRead (Stored payload)
row = (\payload
typed -> JobRead (Stored payload)
row {payload = typed}) (payload -> JobRead payload)
-> Either Text payload -> Either Text (JobRead payload)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Stored payload -> Either Text payload
forall payload.
FromJSON payload =>
Stored payload -> Either Text payload
decodeStored (JobRead (Stored payload) -> Stored payload
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> payload
payload JobRead (Stored payload)
row)
typedRow :: (FromJSON payload, MonadArbiter m) => JobRead (Stored payload) -> m (JobRead payload)
typedRow :: forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
typedRow = Either Text (JobRead payload) -> m (JobRead payload)
forall (m :: * -> *) a. MonadIO m => Either Text a -> m a
orThrowParsing (Either Text (JobRead payload) -> m (JobRead payload))
-> (JobRead (Stored payload) -> Either Text (JobRead payload))
-> JobRead (Stored payload)
-> m (JobRead payload)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. JobRead (Stored payload) -> Either Text (JobRead payload)
forall payload.
FromJSON payload =>
JobRead (Stored payload) -> Either Text (JobRead payload)
decodeRow
typedDLQRow :: (FromJSON payload, MonadArbiter m) => DLQ.DLQJob (Stored payload) -> m (DLQ.DLQJob payload)
typedDLQRow :: forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
DLQJob (Stored payload) -> m (DLQJob payload)
typedDLQRow DLQJob (Stored payload)
entry = (\JobSnapshot payload
snapshot -> DLQJob (Stored payload)
entry {DLQ.jobSnapshot = snapshot}) (JobSnapshot payload -> DLQJob payload)
-> m (JobSnapshot payload) -> m (DLQJob payload)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> JobRead (Stored payload) -> m (JobSnapshot payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
typedRow (DLQJob (Stored payload) -> JobRead (Stored payload)
forall payload. DLQJob payload -> JobSnapshot payload
DLQ.jobSnapshot DLQJob (Stored payload)
entry)
typedArchiveRow
:: (FromJSON payload, MonadArbiter m) => Archive.ArchiveJob (Stored payload) -> m (Archive.ArchiveJob payload)
typedArchiveRow :: forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
ArchiveJob (Stored payload) -> m (ArchiveJob payload)
typedArchiveRow ArchiveJob (Stored payload)
entry = (\JobSnapshot payload
snapshot -> ArchiveJob (Stored payload)
entry {Archive.jobSnapshot = snapshot}) (JobSnapshot payload -> ArchiveJob payload)
-> m (JobSnapshot payload) -> m (ArchiveJob payload)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> JobRead (Stored payload) -> m (JobSnapshot payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
typedRow (ArchiveJob (Stored payload) -> JobRead (Stored payload)
forall payload. ArchiveJob payload -> JobSnapshot payload
Archive.jobSnapshot ArchiveJob (Stored payload)
entry)
type RejectedRow payload = (JobRead (Stored payload), Text)
decodeClaimed
:: forall payload
. (JobPayload payload)
=> [JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload])
decodeClaimed :: forall payload.
JobPayload payload =>
[JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload])
decodeClaimed = ([RejectedRow payload], [JobRead payload])
-> ([JobRead payload], [RejectedRow payload])
forall a b. (a, b) -> (b, a)
swap (([RejectedRow payload], [JobRead payload])
-> ([JobRead payload], [RejectedRow payload]))
-> ([JobRead (Stored Value)]
-> ([RejectedRow payload], [JobRead payload]))
-> [JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload])
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Either (RejectedRow payload) (JobRead payload)]
-> ([RejectedRow payload], [JobRead payload])
forall a b. [Either a b] -> ([a], [b])
partitionEithers ([Either (RejectedRow payload) (JobRead payload)]
-> ([RejectedRow payload], [JobRead payload]))
-> ([JobRead (Stored Value)]
-> [Either (RejectedRow payload) (JobRead payload)])
-> [JobRead (Stored Value)]
-> ([RejectedRow payload], [JobRead payload])
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead (Stored Value)
-> Either (RejectedRow payload) (JobRead payload))
-> [JobRead (Stored Value)]
-> [Either (RejectedRow payload) (JobRead payload)]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead (Stored payload)
-> Either (RejectedRow payload) (JobRead payload)
forall {payload}.
FromJSON payload =>
JobRead (Stored payload)
-> Either (JobRead (Stored payload), Text) (JobRead payload)
decodeOne (JobRead (Stored payload)
-> Either (RejectedRow payload) (JobRead payload))
-> (JobRead (Stored Value) -> JobRead (Stored payload))
-> JobRead (Stored Value)
-> Either (RejectedRow payload) (JobRead payload)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Stored Value -> Stored payload)
-> JobRead (Stored Value) -> JobRead (Stored payload)
forall payload payload' key q insertedAt adm.
(payload -> payload')
-> Job payload key q insertedAt adm
-> Job payload' key q insertedAt adm
mapPayload Stored Value -> Stored payload
forall a b. Coercible a b => a -> b
coerce)
where
decodeOne :: JobRead (Stored payload)
-> Either (JobRead (Stored payload), Text) (JobRead payload)
decodeOne JobRead (Stored payload)
row = (Text -> (JobRead (Stored payload), Text))
-> Either Text (JobRead payload)
-> Either (JobRead (Stored payload), Text) (JobRead payload)
forall a b c. (a -> b) -> Either a c -> Either b c
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first (JobRead (Stored payload)
row,) (JobRead (Stored payload) -> Either Text (JobRead payload)
forall payload.
FromJSON payload =>
JobRead (Stored payload) -> Either Text (JobRead payload)
decodeRow JobRead (Stored payload)
row)
deadLetterRejected :: (MonadArbiter m) => RejectedRow payload -> m Int64
deadLetterRejected :: forall (m :: * -> *) payload.
MonadArbiter m =>
RejectedRow payload -> m Int64
deadLetterRejected (JobRead (Stored payload)
row, Text
err) = do
schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
MA.getSchema
moveToDLQ TakeLocks schemaName (JT.queueName row) err row
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
queryOne :: (MonadArbiter m) => Text -> Q.Query a -> m a
queryOne :: forall (m :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne Text
label Query a
query = do
rows <- Query a -> m [a]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery Query a
query
case rows of
[a
row] -> a -> m a
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure a
row
[a]
_ -> Text -> m a
forall (m :: * -> *) a. MonadIO m => Text -> m a
throwParsing (Text -> m a) -> Text -> m a
forall a b. (a -> b) -> a -> b
$ Text
label Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
": unexpected result"
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
-> Selector RateLimitPolicy payload (Maybe RateLimitKey)
-> Maybe RateLimitKey
forall policy payload a. payload -> Selector policy payload a -> a
runSelector payload
payloadValue (forall payload. HasRateLimit payload => RateLimitFor payload
rateLimitFor @payload)
ccKey :: Maybe ConcurrencyKey
ccKey = payload
-> Selector ConcurrencyPolicy payload (Maybe ConcurrencyKey)
-> Maybe ConcurrencyKey
forall policy payload a. payload -> Selector policy payload a -> a
runSelector 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
}
payloadWrite :: (JobPayload payload) => payload -> (Value, PayloadColumns)
payloadWrite :: forall payload.
JobPayload payload =>
payload -> (Value, PayloadColumns)
payloadWrite payload
value = (payload -> Value
forall a. ToJSON a => a -> Value
toJSON payload
value, payload -> PayloadColumns
forall payload.
(HasConcurrency payload, HasKind payload, HasRateLimit payload) =>
payload -> PayloadColumns
payloadColumns payload
value)
type TraceStamp payload = JobWrite payload -> JobWrite payload
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
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
(Value
encoded, PayloadColumns
derived) = payload -> (Value, PayloadColumns)
forall payload.
JobPayload payload =>
payload -> (Value, PayloadColumns)
payloadWrite (JobWrite payload -> payload
forall payload key q insertedAt adm.
Job 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 = PayloadColumns
derived
, sourceParentId :: Maybe Int64
sourceParentId = Maybe Int64
parent
, sourceParentState :: Maybe Value
sourceParentState = Maybe Value
state
, sourceSuspended :: Bool
sourceSuspended = Bool
suspended
}
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 -> JobWriteSource payload -> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> JobWriteSource payload -> m (Maybe (JobRead payload))
insertJobSource Text
schemaName Text
tableName (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
. (MonadArbiter m)
=> SchemaName
-> TableName
-> JobWriteSource payload
-> m (Maybe (JobRead payload))
insertJobSource :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> JobWriteSource payload -> m (Maybe (JobRead payload))
insertJobSource Text
schemaName Text
tableName JobWriteSource payload
source = do
let job :: JobWrite payload
job = JobWriteSource payload -> JobWrite payload
forall payload. JobWriteSource payload -> JobWrite payload
sourceJob JobWriteSource payload
source
valuesFrag :: Query ()
valuesFrag = Codec (JobWriteSource payload) (JobRead (Stored (ZonkAny 1)))
-> JobWriteSource payload -> Query ()
forall s a. Codec s a -> s -> Query ()
insertFrag (Text
-> Codec (JobWriteSource payload) (JobRead (Stored (ZonkAny 1)))
forall payload stored.
Text -> Codec (JobWriteSource payload) (JobRead (Stored stored))
jobCodec Text
tableName) JobWriteSource payload
source
query :: Query (JobRead (Stored (ZonkAny 2)))
query
| Maybe DedupKey -> Bool
isReplaceDuplicate (JobWrite payload -> Maybe DedupKey
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey JobWrite payload
job) = Text -> Text -> Query () -> Query (JobRead (Stored (ZonkAny 2)))
forall payload.
Text -> Text -> Query () -> Query (JobRead (Stored payload))
Tmpl.insertJobReplaceSQL Text
schemaName Text
tableName Query ()
valuesFrag
| Bool
otherwise = Text -> Text -> Query () -> Query (JobRead (Stored (ZonkAny 2)))
forall payload.
Text -> Text -> Query () -> Query (JobRead (Stored payload))
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
$
Query (JobRead (Stored (ZonkAny 2)))
-> m [JobRead (Stored (ZonkAny 2))]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery Query (JobRead (Stored (ZonkAny 2)))
query m [JobRead (Stored (ZonkAny 2))]
-> ([JobRead (Stored (ZonkAny 2))] -> 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
>>= \case
JobRead (Stored (ZonkAny 2))
raw : [JobRead (Stored (ZonkAny 2))]
_ -> Maybe (JobRead payload) -> m (Maybe (JobRead payload))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (JobRead payload -> Maybe (JobRead payload)
forall a. a -> Maybe a
Just JobRead (Stored (ZonkAny 2))
raw {payload = JT.payload job})
[]
| Maybe DedupKey -> Bool
forall a. Maybe a -> Bool
isJust (JobWrite payload -> Maybe DedupKey
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey JobWrite payload
job) -> 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
| Bool
otherwise -> Text -> m (Maybe (JobRead payload))
forall (m :: * -> *) a. MonadIO m => Text -> m a
throwParsing Text
"insertJob: No rows returned from INSERT"
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 =
Text -> Text -> Query () -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Query () -> m [JobRead payload]
insertBatchSource Text
schemaName Text
tableName (Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
forall payload.
JobPayload payload =>
Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
batchSource Text
tableName TraceStamp payload
stamp (Int64 -> Maybe Int64
forall a. a -> Maybe a
Just Int64
parent) [JobWrite payload]
jobs)
batchSource
:: (JobPayload payload)
=> TableName
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Q.Query ()
batchSource :: forall payload.
JobPayload payload =>
Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
batchSource Text
tableName TraceStamp payload
stamp Maybe Int64
parent = Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
forall payload.
JobPayload payload =>
Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
stampedRowsSource Text
tableName TraceStamp payload
stamp Maybe Int64
parent ([JobWrite payload] -> Query ())
-> ([JobWrite payload] -> [JobWrite payload])
-> [JobWrite payload]
-> Query ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [JobWrite payload] -> [JobWrite payload]
forall payload. [JobWrite payload] -> [JobWrite payload]
dedupBatch
stampedRowsSource
:: (JobPayload payload)
=> TableName
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Q.Query ()
stampedRowsSource :: forall payload.
JobPayload payload =>
Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
stampedRowsSource Text
tableName TraceStamp payload
stamp Maybe Int64
parent [JobWrite payload]
jobs =
Codec (JobWriteSource payload) (JobRead (Stored (ZonkAny 3)))
-> [JobWriteSource payload] -> Query ()
forall s a. Codec s a -> [s] -> Query ()
batchFrag (Text
-> Codec (JobWriteSource payload) (JobRead (Stored (ZonkAny 3)))
forall payload stored.
Text -> Codec (JobWriteSource payload) (JobRead (Stored stored))
jobCodec Text
tableName) [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
forall a. Maybe a
Nothing Bool
False JobWrite payload
job | JobWrite payload
job <- [JobWrite payload]
jobs]
insertBatchSource
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> SchemaName
-> TableName
-> Q.Query ()
-> m [JobRead payload]
insertBatchSource :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Query () -> m [JobRead payload]
insertBatchSource Text
schemaName Text
tableName Query ()
batchSrc =
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
$
Query (JobRead (Stored payload)) -> m [JobRead (Stored payload)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query () -> Query (JobRead (Stored payload))
forall payload.
Text -> Text -> Query () -> Query (JobRead (Stored payload))
Tmpl.insertJobsBatchSQL Text
schemaName Text
tableName Query ()
batchSrc) m [JobRead (Stored payload)]
-> ([JobRead (Stored 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
>>= (JobRead (Stored payload) -> m (JobRead payload))
-> [JobRead (Stored payload)] -> m [JobRead payload]
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 JobRead (Stored payload) -> m (JobRead payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
typedRow
spawnChildren
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> SchemaName
-> TableName
-> JobRead payload
-> NonEmpty (JobWrite payload)
-> m (NonEmpty (JobRead payload))
spawnChildren :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> JobRead payload
-> NonEmpty (JobWrite payload)
-> m (NonEmpty (JobRead payload))
spawnChildren Text
schemaName Text
tableName JobRead payload
job NonEmpty (JobWrite payload)
children = 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
Text -> Text -> [(Int64, Maybe Int64)] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [(Int64, Maybe Int64)] -> m ()
lockJobsAndParents Text
schemaName Text
tableName [(Int64
jobId, JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe Int64
parentId JobRead payload
job)]
marked <- Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Text -> Int64 -> Int64 -> Query ()
Tmpl.beginSpawnSQL Text
schemaName Text
tableName Int64
jobId (JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job))
when (marked == 0) refuseSpawn
when (Set.size groups > 1)
$ void
$ MA.executeStatement (Tmpl.lockGroupKeysSQL schemaName tableName (Set.toAscList groups))
when (isRollup job) $ do
void $ MA.executeStatement (Tmpl.deleteResultsByParentSQL schemaName tableName jobId)
void $ MA.executeStatement (Tmpl.detachDLQChildrenSQL schemaName tableName jobId)
stamp <- traceStamp
let writes = [JobWrite payload] -> [JobWrite payload]
forall payload. [JobWrite payload] -> [JobWrite payload]
dedupBatch (NonEmpty (JobWrite payload) -> [JobWrite payload]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty (JobWrite payload)
children)
inserted <- insertBatchSource schemaName tableName (stampedRowsSource tableName stamp (Just jobId) writes)
let missing = [JobWrite payload] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [JobWrite payload]
writes Int -> Int -> Int
forall a. Num a => a -> a -> a
- [JobRead payload] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [JobRead payload]
inserted
case NE.nonEmpty inserted of
Just NonEmpty (JobRead payload)
spawned | Int
missing Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 -> NonEmpty (JobRead payload) -> m (NonEmpty (JobRead payload))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure NonEmpty (JobRead payload)
spawned
Maybe (NonEmpty (JobRead payload))
_ ->
Text -> m (NonEmpty (JobRead payload))
forall a. Text -> m a
scopedRetry (Text -> m (NonEmpty (JobRead payload)))
-> Text -> m (NonEmpty (JobRead payload))
forall a b. (a -> b) -> a -> b
$
Text
"spawnChildren: "
Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall a. Show a => a -> Text
showText Int
missing
Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" of "
Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall a. Show a => a -> Text
showText ([JobWrite payload] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [JobWrite payload]
writes)
Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" children of job "
Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int64 -> Text
forall a. Show a => a -> Text
showText Int64
jobId
Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" hit a dedup conflict"
where
jobId :: Int64
jobId = JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job
groups :: Set Text
groups = [Text] -> Set Text
forall a. Ord a => [a] -> Set a
Set.fromList ([Maybe Text] -> [Text]
forall a. [Maybe a] -> [a]
catMaybes (JobRead payload -> Maybe Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe Text
groupKey JobRead payload
job Maybe Text -> [Maybe Text] -> [Maybe Text]
forall a. a -> [a] -> [a]
: (JobWrite payload -> Maybe Text)
-> [JobWrite payload] -> [Maybe Text]
forall a b. (a -> b) -> [a] -> [b]
map JobWrite payload -> Maybe Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe Text
JT.groupKey (NonEmpty (JobWrite payload) -> [JobWrite payload]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty (JobWrite payload)
children)))
scopedRetry :: forall a. Text -> m a
scopedRetry :: forall a. Text -> m a
scopedRetry Text
msg = JobException -> [Int64] -> m a
forall (m :: * -> *) a. MonadIO m => JobException -> [Int64] -> m a
throwScopedFailure (JobRetryableException -> JobException
Retryable (Text -> Maybe NominalDiffTime -> JobRetryableException
JobRetryableException Text
msg Maybe NominalDiffTime
forall a. Maybe a
Nothing)) [Int64
jobId]
refuseSpawn :: m ()
refuseSpawn :: m ()
refuseSpawn = do
alreadySpawned <- Query Bool -> m Bool
forall (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> Text -> Int64 -> Int64 -> Query Bool
Tmpl.spawnedAlreadySQL Text
schemaName Text
tableName Int64
jobId (JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job))
if alreadySpawned && isJust (claimedBy job)
then
scopedRetry $
"spawnChildren: job " <> showText jobId <> " already spawned children in this round"
else throwJobGoneIds "no longer claimed by this worker" [jobId]
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
rlkPrefix RateLimitKey
key) (RateLimitKey -> Text
rateLimitKeyText RateLimitKey
key) Double
amount)
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))
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)
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 :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne Text
"wakeThrottledJobs" (Text -> [Text] -> Text -> Query Int64
Tmpl.wakeThrottledJobsSQL Text
schemaName [Text]
tableNames Text
prefix)
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 :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne
Text
"wakeThrottledJobsForKey"
(Text -> [Text] -> Text -> Text -> Query Int64
Tmpl.wakeThrottledJobsForKeySQL Text
schemaName [Text]
tableNames (RateLimitKey -> Text
rlkPrefix RateLimitKey
key) (RateLimitKey -> Text
rateLimitKeyText RateLimitKey
key))
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] -> Maybe Text -> Query RateLimitPolicyView
Tmpl.rateLimitPoliciesSQL Text
schemaName [Text]
tableNames Maybe Text
forall a. Maybe a
Nothing)
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 =
Query RateLimitPolicyView -> m (Maybe RateLimitPolicyView)
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> [Text] -> Maybe Text -> Query RateLimitPolicyView
Tmpl.rateLimitPoliciesSQL Text
schemaName [Text]
tableNames (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
prefix))
rateLimitPolicyExists :: (MonadArbiter m) => SchemaName -> Text -> m Bool
rateLimitPolicyExists :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Bool
rateLimitPolicyExists Text
schemaName Text
prefix =
Query Bool -> m Bool
forall (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> Text -> Query Bool
Tmpl.rateLimitPolicyExistsSQL Text
schemaName Text
prefix)
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))
updateRateLimitPolicyOverrides :: (MonadArbiter m) => SchemaName -> Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides Text
schemaName Text
prefix RateLimitPolicyUpdate {overrideMaxTokens :: RateLimitPolicyUpdate -> Maybe (Maybe Double)
overrideMaxTokens = Maybe (Maybe Double)
mMax, overrideRefillAmount :: RateLimitPolicyUpdate -> Maybe (Maybe Double)
overrideRefillAmount = Maybe (Maybe Double)
mRefill, overrideInterval :: RateLimitPolicyUpdate -> Maybe (Maybe Double)
overrideInterval = 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.updateRateLimitPolicyOverridesSQL Text
schemaName Maybe (Maybe Double)
mMax Maybe (Maybe Double)
mRefill Maybe (Maybe Double)
mInterval Text
prefix)
updateConcurrencyPolicyOverrides :: (MonadArbiter m) => SchemaName -> Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides Text
schemaName Text
prefix ConcurrencyPolicyUpdate {overrideLimit :: ConcurrencyPolicyUpdate -> Maybe (Maybe Int32)
overrideLimit = 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.updateConcurrencyPolicyOverridesSQL Text
schemaName Maybe (Maybe Int32)
mLim Text
prefix)
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 -> Maybe Text -> Query ConcurrencyPolicyView
Tmpl.concurrencyPoliciesSQL Text
schemaName Maybe Text
forall a. Maybe a
Nothing)
getConcurrencyPolicy :: (MonadArbiter m) => SchemaName -> Text -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy Text
schemaName Text
prefix =
Query ConcurrencyPolicyView -> m (Maybe ConcurrencyPolicyView)
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Maybe Text -> Query ConcurrencyPolicyView
Tmpl.concurrencyPoliciesSQL Text
schemaName (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
prefix))
concurrencyPolicyExists :: (MonadArbiter m) => SchemaName -> Text -> m Bool
concurrencyPolicyExists :: forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Bool
concurrencyPolicyExists Text
schemaName Text
prefix =
Query Bool -> m Bool
forall (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> Text -> Query Bool
Tmpl.concurrencyPolicyExistsSQL Text
schemaName Text
prefix)
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))
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)
locked <- if null dead then pure [] else MA.executeQuery (Tmpl.tryLockDeadConcurrencyAdvisorySQL dead)
if null locked
then pure 0
else MA.executeStatement (Tmpl.pruneLockedConcurrencyKeysSQL schemaName tableNames locked)
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)
countOr0 (Tmpl.reconcileConcurrencyCountsSQL schemaName tableNames held)
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 (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> [Text] -> Query Bool
Tmpl.concurrencyCountsStaleSQL Text
schemaName [Text]
tableNames)
if stale then reconcileConcurrencyCounts schemaName tableNames else pure 0
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 (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> Query Bool
Tmpl.concurrencyHasAnyKeySQL Text
schemaName)
if hasKeys
then
(+)
<$> reconcileConcurrencyCounts schemaName tableNames
<*> pruneConcurrencyKeys schemaName tableNames
else pure 0
insertJob
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> SchemaName
-> TableName
-> 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 -> JobWriteSource payload -> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> JobWriteSource payload -> m (Maybe (JobRead payload))
insertJobSource Text
schemaName Text
tableName (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)
insertJobsBatch
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> SchemaName
-> TableName
-> [JobWrite payload]
-> 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 -> Query () -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Query () -> m [JobRead payload]
insertBatchSource Text
schemaName Text
tableName (Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
forall payload.
JobPayload payload =>
Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
batchSource Text
tableName TraceStamp payload
stamp Maybe Int64
forall a. Maybe a
Nothing [JobWrite payload]
jobs)
insertJobsBatch_
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> SchemaName
-> TableName
-> [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 =
m (TraceStamp payload)
forall (m :: * -> *) payload. MonadIO m => m (TraceStamp payload)
traceStamp m (TraceStamp payload)
-> (TraceStamp payload -> m Int64) -> m Int64
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 ->
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
$
Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Text -> Query () -> Query ()
Tmpl.insertJobsBatchSQL_ Text
schemaName Text
tableName (Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
forall payload.
JobPayload payload =>
Text
-> TraceStamp payload
-> Maybe Int64
-> [JobWrite payload]
-> Query ()
batchSource Text
tableName TraceStamp payload
stamp Maybe Int64
forall a. Maybe a
Nothing [JobWrite payload]
jobs))
insertResult
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> Int64
-> 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)
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)
getResultsByParent
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Map Int64 Value)
getResultsByParent :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Map Int64 Value)
getResultsByParent Text
schemaName Text
tableName Int64
parentJobId =
[(Int64, Value)] -> Map Int64 Value
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(Int64, Value)] -> Map Int64 Value)
-> m [(Int64, Value)] -> m (Map Int64 Value)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> 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)
getDLQChildErrorsByParent
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Map Int64 Text)
getDLQChildErrorsByParent :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Map Int64 Text)
getDLQChildErrorsByParent Text
schemaName Text
tableName Int64
parentJobId =
[(Int64, Text)] -> Map Int64 Text
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(Int64, Text)] -> Map Int64 Text)
-> ([(Int64, Maybe Text)] -> [(Int64, Text)])
-> [(Int64, Maybe Text)]
-> Map Int64 Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ((Int64, Maybe Text) -> Maybe (Int64, Text))
-> [(Int64, Maybe Text)] -> [(Int64, Text)]
forall a b. (a -> Maybe b) -> [a] -> [b]
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)
([(Int64, Maybe Text)] -> Map Int64 Text)
-> m [(Int64, Maybe Text)] -> m (Map Int64 Text)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> 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)
persistParentState
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> Value
-> 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)
getParentStateSnapshot
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Maybe Value)
getParentStateSnapshot :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe Value)
getParentStateSnapshot Text
schemaName Text
tableName Int64
jobId =
Maybe (Maybe Value) -> Maybe Value
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join (Maybe (Maybe Value) -> Maybe Value)
-> m (Maybe (Maybe Value)) -> m (Maybe Value)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query (Maybe Value) -> m (Maybe (Maybe Value))
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Text -> Int64 -> Query (Maybe Value)
Tmpl.getParentStateSnapshotSQL Text
schemaName Text
tableName Int64
jobId)
readChildResultsRaw
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Map Int64 Value, Map Int64 Text, Maybe Value, 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
_, Maybe a
mErr, Just k
dlqPk) ->
let err :: a
err = a -> Maybe a -> a
forall a. a -> Maybe a -> a
fromMaybe a
"" Maybe a
mErr
in (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
"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
<> (a, Maybe k, Maybe a, Maybe a, Maybe k) -> Text
forall a. Show a => a -> Text
showText (a, Maybe k, Maybe a, Maybe a, Maybe k)
row
mergeRawChildResults
:: Map Int64 Value
-> Map Int64 Text
-> Maybe Value
-> 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 = (Value -> Map Int64 (Either Text Value))
-> Maybe Value -> Map Int64 (Either Text Value)
forall m a. Monoid m => (a -> m) -> Maybe a -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap Value -> Map Int64 (Either Text Value)
forall a. (FromJSON a, Monoid a) => Value -> a
decodeOrEmpty Maybe Value
mSnapshot
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 (Job payload key q insertedAt adm))
-> Job payload key q insertedAt adm
-> (Map Text Int, Seq (Job 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 (Job payload key q insertedAt adm))
-> Job payload key q insertedAt adm
-> (Map Text Int, Seq (Job payload key q insertedAt adm))
step (!Map Text Int
seen, !Seq (Job payload key q insertedAt adm)
rows) Job payload key q insertedAt adm
job = case (Maybe Text, Maybe Text) -> Maybe Text
forall a b. (a, b) -> a
fst (Maybe DedupKey -> (Maybe Text, Maybe Text)
dedupParts (Job payload key q insertedAt adm -> Maybe DedupKey
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey Job payload key q insertedAt adm
job)) of
Maybe Text
Nothing -> (Map Text Int
seen, Seq (Job payload key q insertedAt adm)
rows Seq (Job payload key q insertedAt adm)
-> Job payload key q insertedAt adm
-> Seq (Job payload key q insertedAt adm)
forall a. Seq a -> a -> Seq a
|> Job 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 (Job payload key q insertedAt adm) -> Int
forall a. Seq a -> Int
Seq.length Seq (Job payload key q insertedAt adm)
rows) Map Text Int
seen, Seq (Job payload key q insertedAt adm)
rows Seq (Job payload key q insertedAt adm)
-> Job payload key q insertedAt adm
-> Seq (Job payload key q insertedAt adm)
forall a. Seq a -> a -> Seq a
|> Job payload key q insertedAt adm
job)
Just Int
idx
| Maybe DedupKey -> Bool
isReplaceDuplicate (Job payload key q insertedAt adm -> Maybe DedupKey
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe DedupKey
JT.dedupKey Job payload key q insertedAt adm
job) -> (Map Text Int
seen, Int
-> Job payload key q insertedAt adm
-> Seq (Job payload key q insertedAt adm)
-> Seq (Job payload key q insertedAt adm)
forall a. Int -> a -> Seq a -> Seq a
Seq.update Int
idx Job payload key q insertedAt adm
job Seq (Job payload key q insertedAt adm)
rows)
| Bool
otherwise -> (Map Text Int
seen, Seq (Job payload key q insertedAt adm)
rows)
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 RateLimitPolicy 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)
}
anonymousClaimant :: UUID
anonymousClaimant :: UUID
anonymousClaimant = Word32 -> Word32 -> Word32 -> Word32 -> UUID
UUID.fromWords Word32
0xa4b17e40 Word32
0 Word32
0 Word32
1
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]
claimNextVisibleJobsAs Text
schemaName Text
tableName Int
maxJobs NominalDiffTime
timeout UUID
anonymousClaimant
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 =
JobStatements
-> Int -> m ([JobRead payload], [RejectedRow payload])
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
JobStatements
-> Int -> m ([JobRead payload], [RejectedRow payload])
claimJobsCached
(forall payload.
JobPayload payload =>
Text
-> Text -> Int -> Int -> NominalDiffTime -> UUID -> JobStatements
mkJobStatements @payload Text
schemaName Text
tableName Int
singleJobBatchSize Int
noStagedCapacities NominalDiffTime
timeout UUID
workerId)
Int
maxJobs
m ([JobRead payload], [RejectedRow payload])
-> (([JobRead payload], [RejectedRow 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
>>= ([JobRead payload], [RejectedRow payload]) -> m [JobRead payload]
forall (m :: * -> *) a payload.
MonadArbiter m =>
(a, [RejectedRow payload]) -> m a
deadLetterRest
claimJobsCached
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> JobStatements
-> Int
-> m ([JobRead payload], [RejectedRow payload])
claimJobsCached :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
JobStatements
-> Int -> m ([JobRead payload], [RejectedRow payload])
claimJobsCached JobStatements
statements Int
maxJobs =
[JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload])
forall payload.
JobPayload payload =>
[JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload])
decodeClaimed ([JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload]))
-> m [JobRead (Stored Value)]
-> m ([JobRead payload], [RejectedRow payload])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query (JobRead (Stored Value)) -> m [JobRead (Stored Value)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQueryPrepared (JobStatements -> Int -> Query (JobRead (Stored Value))
claimFor JobStatements
statements Int
maxJobs)
deadLetterRest :: (MonadArbiter m) => (a, [RejectedRow payload]) -> m a
deadLetterRest :: forall (m :: * -> *) a payload.
MonadArbiter m =>
(a, [RejectedRow payload]) -> m a
deadLetterRest (a
jobs, [RejectedRow payload]
rejected) = a
jobs a -> m () -> m a
forall a b. a -> m b -> m a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ (RejectedRow payload -> m (Either SomeException Int64))
-> [RejectedRow payload] -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (m Int64 -> m (Either SomeException Int64)
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> m (Either SomeException a)
tryAny (m Int64 -> m (Either SomeException Int64))
-> (RejectedRow payload -> m Int64)
-> RejectedRow payload
-> m (Either SomeException Int64)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. RejectedRow payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
RejectedRow payload -> m Int64
deadLetterRejected) [RejectedRow payload]
rejected
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 =
JobStatements
-> Int -> m ([NonEmpty (JobRead payload)], [RejectedRow payload])
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
JobStatements
-> Int -> m ([NonEmpty (JobRead payload)], [RejectedRow payload])
claimJobsBatchedCached
(forall payload.
JobPayload payload =>
Text
-> Text -> Int -> Int -> NominalDiffTime -> UUID -> JobStatements
mkJobStatements @payload Text
schemaName Text
tableName Int
batchSize Int
noStagedCapacities NominalDiffTime
timeout UUID
anonymousClaimant)
Int
maxBatches
m ([NonEmpty (JobRead payload)], [RejectedRow payload])
-> (([NonEmpty (JobRead payload)], [RejectedRow payload])
-> m [NonEmpty (JobRead payload)])
-> m [NonEmpty (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
>>= ([NonEmpty (JobRead payload)], [RejectedRow payload])
-> m [NonEmpty (JobRead payload)]
forall (m :: * -> *) a payload.
MonadArbiter m =>
(a, [RejectedRow payload]) -> m a
deadLetterRest
claimJobsBatchedCached
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> JobStatements
-> Int
-> m ([NonEmpty (JobRead payload)], [RejectedRow payload])
claimJobsBatchedCached :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
JobStatements
-> Int -> m ([NonEmpty (JobRead payload)], [RejectedRow payload])
claimJobsBatchedCached JobStatements
statements Int
maxBatches
| JobStatements -> Int
claimBatchSize JobStatements
statements Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
1 = ([NonEmpty (JobRead payload)], [RejectedRow payload])
-> m ([NonEmpty (JobRead payload)], [RejectedRow 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)], [RejectedRow payload])
-> m ([NonEmpty (JobRead payload)], [RejectedRow payload])
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([], [])
| Bool
otherwise = do
(jobs, rejected) <- [JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload])
forall payload.
JobPayload payload =>
[JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload])
decodeClaimed ([JobRead (Stored Value)]
-> ([JobRead payload], [RejectedRow payload]))
-> m [JobRead (Stored Value)]
-> m ([JobRead payload], [RejectedRow payload])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query (JobRead (Stored Value)) -> m [JobRead (Stored Value)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQueryPrepared (JobStatements -> Int -> Query (JobRead (Stored Value))
claimFor JobStatements
statements Int
maxBatches)
pure (concatMap (chunksOfNE (claimBatchSize statements)) (NE.groupAllWith groupKey jobs), rejected)
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
archivesOnAck :: JobRead payload -> Bool
archivesOnAck :: forall payload. JobRead payload -> Bool
archivesOnAck = (Int32 -> Bool) -> Maybe Int32 -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (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.
Job payload key q insertedAt adm -> Maybe Int32
archiveFor
ackJob
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> JobRead payload
-> m Int64
ackJob :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> JobRead payload -> m Int64
ackJob 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
. AckStatements -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
AckStatements -> JobRead payload -> m Int64
ackJobWith (Text -> Text -> AckStatements
mkAckStatements Text
schemaName Text
tableName)
data AckStatements = AckStatements
{ AckStatements -> Text
ackSchema :: SchemaName
, AckStatements -> Text
ackTable :: TableName
, AckStatements -> Bool -> Int64 -> Int64 -> Query Int64
ackFor :: Bool -> Int64 -> Int64 -> Q.Query Int64
, AckStatements -> Bool -> [Int64] -> [Int64] -> Query Int64
ackBatchFor :: Bool -> [Int64] -> [Int64] -> Q.Query Int64
}
mkAckStatements :: SchemaName -> TableName -> AckStatements
mkAckStatements :: Text -> Text -> AckStatements
mkAckStatements Text
schemaName Text
tableName =
AckStatements
{ ackSchema :: Text
ackSchema = Text
schemaName
, ackTable :: Text
ackTable = Text
tableName
, ackFor :: Bool -> Int64 -> Int64 -> Query Int64
ackFor = \Bool
archiving -> if Bool
archiving then Int64 -> Int64 -> Query Int64
ackArchiving else Int64 -> Int64 -> Query Int64
ackPlain
, ackBatchFor :: Bool -> [Int64] -> [Int64] -> Query Int64
ackBatchFor = \Bool
archiving -> if Bool
archiving then [Int64] -> [Int64] -> Query Int64
ackBatchArchiving else [Int64] -> [Int64] -> Query Int64
ackBatchPlain
}
where
ackPlain :: Int64 -> Int64 -> Query Int64
ackPlain = Bool -> Text -> Text -> Int64 -> Int64 -> Query Int64
Tmpl.smartAckJobSQL Bool
False Text
schemaName Text
tableName
ackArchiving :: Int64 -> Int64 -> Query Int64
ackArchiving = Bool -> Text -> Text -> Int64 -> Int64 -> Query Int64
Tmpl.smartAckJobSQL Bool
True Text
schemaName Text
tableName
ackBatchPlain :: [Int64] -> [Int64] -> Query Int64
ackBatchPlain = Bool -> Text -> Text -> [Int64] -> [Int64] -> Query Int64
Tmpl.smartAckJobsBatchSQL Bool
False Text
schemaName Text
tableName
ackBatchArchiving :: [Int64] -> [Int64] -> Query Int64
ackBatchArchiving = Bool -> Text -> Text -> [Int64] -> [Int64] -> Query Int64
Tmpl.smartAckJobsBatchSQL Bool
True Text
schemaName Text
tableName
data JobStatements = JobStatements
{ JobStatements -> Int
claimBatchSize :: Int
, JobStatements -> Int -> Query (JobRead (Stored Value))
claimFor :: Int -> Q.Query (JobRead (Stored Value))
, JobStatements -> AckStatements
statementsAck :: AckStatements
}
mkJobStatements
:: forall payload
. (JobPayload payload)
=> SchemaName
-> TableName
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> JobStatements
mkJobStatements :: forall payload.
JobPayload payload =>
Text
-> Text -> Int -> Int -> NominalDiffTime -> UUID -> JobStatements
mkJobStatements Text
schemaName Text
tableName Int
batchSize Int
poolSize NominalDiffTime
timeout UUID
workerId =
JobStatements
{ claimBatchSize :: Int
claimBatchSize = Int
batchSize
, claimFor :: Int -> Query (JobRead (Stored Value))
claimFor = \Int
capacity -> Query (JobRead (Stored Value))
-> Int
-> IntMap (Query (JobRead (Stored Value)))
-> Query (JobRead (Stored Value))
forall a. a -> Int -> IntMap a -> a
IntMap.findWithDefault (Int -> Query (JobRead (Stored Value))
renderClaim Int
capacity) Int
capacity IntMap (Query (JobRead (Stored Value)))
claims
, statementsAck :: AckStatements
statementsAck = Text -> Text -> AckStatements
mkAckStatements Text
schemaName Text
tableName
}
where
admission :: ClaimAdmission
admission = forall payload.
(HasConcurrency payload, HasRateLimit payload) =>
ClaimAdmission
claimAdmissionFor @payload
codec :: RowCodec (JobRead (Stored Value))
codec = Text -> RowCodec (JobRead (Stored Value))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName
renderClaim :: Int -> Query (JobRead (Stored Value))
renderClaim Int
capacity =
RowCodec (JobRead (Stored Value))
-> Query () -> Query (JobRead (Stored Value))
forall a. RowCodec a -> Query () -> Query a
Q.rows RowCodec (JobRead (Stored Value))
codec (Text
-> Text
-> ClaimAdmission
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> Query ()
Claim.claimJobsBatchedSQL Text
schemaName Text
tableName ClaimAdmission
admission Int
batchSize Int
capacity NominalDiffTime
timeout UUID
workerId)
claims :: IntMap (Query (JobRead (Stored Value)))
claims = [(Int, Query (JobRead (Stored Value)))]
-> IntMap (Query (JobRead (Stored Value)))
forall a. [(Int, a)] -> IntMap a
IntMap.fromList [(Int
capacity, Int -> Query (JobRead (Stored Value))
renderClaim Int
capacity) | Int
capacity <- [Int
1 .. Int
poolSize]]
singleJobBatchSize :: Int
singleJobBatchSize :: Int
singleJobBatchSize = Int
1
noStagedCapacities :: Int
noStagedCapacities :: Int
noStagedCapacities = Int
0
ackJobWith :: (MonadArbiter m) => AckStatements -> JobRead payload -> m Int64
ackJobWith :: forall (m :: * -> *) payload.
MonadArbiter m =>
AckStatements -> JobRead payload -> m Int64
ackJobWith AckStatements
statements JobRead payload
job = do
AckStatements -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
AckStatements -> [Maybe Int64] -> m ()
ackParents AckStatements
statements [JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe Int64
parentId JobRead payload
job]
Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0Prepared (AckStatements -> Bool -> Int64 -> Int64 -> Query Int64
ackFor AckStatements
statements (JobRead payload -> Bool
forall payload. JobRead payload -> Bool
archivesOnAck JobRead payload
job) (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) (JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job))
distinctParents :: [Maybe Int64] -> [Int64]
distinctParents :: [Maybe Int64] -> [Int64]
distinctParents = 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
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.executeQueryPrepared (Text -> [Int64] -> Query (Maybe Text)
Tmpl.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 = [Maybe Int64] -> [Int64]
distinctParents [Maybe Int64]
parents
lockJobRootsAndParents :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m ()
lockJobRootsAndParents :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m ()
lockJobRootsAndParents Text
_ Text
_ [] = () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
lockJobRootsAndParents Text
schemaName Text
tableName [Int64]
roots = do
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]
roots)
lockJobParents schemaName tableName (map Just roots <> parents)
lockJobsAndParents :: (MonadArbiter m) => SchemaName -> TableName -> [(Int64, Maybe Int64)] -> m ()
lockJobsAndParents :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [(Int64, Maybe Int64)] -> m ()
lockJobsAndParents Text
schemaName Text
tableName [(Int64, Maybe Int64)]
pairs =
Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName ([Int64 -> Maybe Int64
forall a. a -> Maybe a
Just Int64
jobId | (Int64
jobId, Maybe Int64
_) <- [(Int64, Maybe Int64)]
pairs] [Maybe Int64] -> [Maybe Int64] -> [Maybe Int64]
forall a. Semigroup a => a -> a -> a
<> ((Int64, Maybe Int64) -> Maybe Int64)
-> [(Int64, Maybe Int64)] -> [Maybe Int64]
forall a b. (a -> b) -> [a] -> [b]
map (Int64, Maybe Int64) -> Maybe Int64
forall a b. (a, b) -> b
snd [(Int64, Maybe Int64)]
pairs)
ackParents :: (MonadArbiter m) => AckStatements -> [Maybe Int64] -> m ()
ackParents :: forall (m :: * -> *).
MonadArbiter m =>
AckStatements -> [Maybe Int64] -> m ()
ackParents AckStatements
statements = Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents (AckStatements -> Text
ackSchema AckStatements
statements) (AckStatements -> Text
ackTable AckStatements
statements)
lockParentAndSelf :: (MonadArbiter m) => SchemaName -> TableName -> Int64 -> m (Maybe Int64)
lockParentAndSelf :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe Int64)
lockParentAndSelf Text
schemaName Text
tableName Int64
jobId = do
mParentId <- Maybe (Maybe Int64) -> Maybe Int64
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join (Maybe (Maybe Int64) -> Maybe Int64)
-> m (Maybe (Maybe Int64)) -> m (Maybe Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query (Maybe Int64) -> m (Maybe (Maybe Int64))
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Text -> [Int64] -> Query (Maybe Int64)
Tmpl.getParentIdsSQL Text
schemaName Text
tableName [Int64
jobId])
mParentId <$ lockJobsAndParents schemaName tableName [(jobId, mParentId)]
resumeJobParents :: (MonadArbiter m) => TreesLocked -> SchemaName -> TableName -> [Maybe Int64] -> m ()
resumeJobParents :: forall (m :: * -> *).
MonadArbiter m =>
TreesLocked -> Text -> Text -> [Maybe Int64] -> m ()
resumeJobParents TreesLocked
held Text
schemaName Text
tableName =
(Int64 -> m ()) -> [Int64] -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (TreesLocked -> Text -> Text -> Int64 -> m ()
forall (m :: * -> *).
MonadArbiter m =>
TreesLocked -> Text -> Text -> Int64 -> m ()
tryResumeParent TreesLocked
held Text
schemaName Text
tableName) ([Int64] -> m ())
-> ([Maybe Int64] -> [Int64]) -> [Maybe Int64] -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Maybe Int64] -> [Int64]
distinctParents
data TreesLocked = TreesLocked
lockJobTrees :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m TreesLocked
lockJobTrees :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockJobTrees Text
_ Text
_ [] = TreesLocked -> m TreesLocked
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure TreesLocked
TreesLocked
lockJobTrees Text
schemaName Text
tableName [Int64]
ids = TreesLocked
TreesLocked TreesLocked -> m Int64 -> m TreesLocked
forall a b. a -> m b -> m a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ 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)
lockParentsAndTrees :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m TreesLocked
lockParentsAndTrees :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockParentsAndTrees Text
schemaName Text
tableName [Int64]
ids = do
Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName ([Maybe Int64] -> m ()) -> m [Maybe Int64] -> m ()
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< 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)
Text -> Text -> [Int64] -> m TreesLocked
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockJobTrees Text
schemaName Text
tableName [Int64]
ids
lockJobTreesFromRoot :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m TreesLocked
lockJobTreesFromRoot :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockJobTreesFromRoot Text
_ Text
_ [] = TreesLocked -> m TreesLocked
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure TreesLocked
TreesLocked
lockJobTreesFromRoot Text
schemaName Text
tableName [Int64]
ids =
TreesLocked
TreesLocked TreesLocked -> m Int64 -> m TreesLocked
forall a b. a -> m b -> m a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ 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)
tryResumeParent :: (MonadArbiter m) => TreesLocked -> SchemaName -> TableName -> Int64 -> m ()
tryResumeParent :: forall (m :: * -> *).
MonadArbiter m =>
TreesLocked -> Text -> Text -> Int64 -> m ()
tryResumeParent TreesLocked
TreesLocked Text
schemaName 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)
ackJobsBatch
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [JobRead payload]
-> m [Int64]
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
. AckStatements -> [JobRead payload] -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
AckStatements -> [JobRead payload] -> m [Int64]
ackJobsBatchWith (Text -> Text -> AckStatements
mkAckStatements Text
schemaName Text
tableName)
ackJobsBatchWith :: (MonadArbiter m) => AckStatements -> [JobRead payload] -> m [Int64]
ackJobsBatchWith :: forall (m :: * -> *) payload.
MonadArbiter m =>
AckStatements -> [JobRead payload] -> m [Int64]
ackJobsBatchWith AckStatements
_ [] = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
ackJobsBatchWith AckStatements
statements [JobRead payload]
jobs = do
AckStatements -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
AckStatements -> [Maybe Int64] -> m ()
ackParents AckStatements
statements ((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.
Job 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 (AckStatements -> Bool -> [Int64] -> [Int64] -> Query Int64
ackBatchFor AckStatements
statements ((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) ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
Job 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.
Job payload Int64 q insertedAt adm -> Int64
claimSeq [JobRead payload]
jobs))
setVisibilityTimeout
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> NominalDiffTime
-> JobRead payload
-> m Int64
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.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) (JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job))
data VisibilityUpdateInfo = VisibilityUpdateInfo
{ VisibilityUpdateInfo -> Int64
vuiJobId :: JobId
, VisibilityUpdateInfo -> Bool
vuiWasHeartbeated :: Bool
, VisibilityUpdateInfo -> Maybe Int64
vuiCurrentDbClaimSeq :: Maybe ClaimSeq
, VisibilityUpdateInfo -> Bool
vuiCancelRequested :: Bool
, VisibilityUpdateInfo -> Bool
vuiSuspended :: Bool
, VisibilityUpdateInfo -> Maybe UUID
vuiClaimedBy :: Maybe UUID
}
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)
setVisibilityTimeoutBatch
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> NominalDiffTime
-> [JobRead payload]
-> m [VisibilityUpdateInfo]
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 =
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 -> [JobRead payload] -> Double -> Query ()
forall payload.
Text -> Text -> [JobRead payload] -> Double -> Query ()
Tmpl.setVisibilityTimeoutBatchSQL Text
schemaName Text
tableName [JobRead payload]
jobs (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac NominalDiffTime
timeout)
updateJobForRetry
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> NominalDiffTime
-> Text
-> 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.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) (JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job))
nackJob
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Maybe NominalDiffTime
-> JobRead payload
-> m Int64
nackJob :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Maybe NominalDiffTime -> JobRead payload -> m Int64
nackJob Text
schemaName Text
tableName Maybe NominalDiffTime
delay 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 -> Maybe Int64 -> Int64 -> Int64 -> Int32 -> Query ()
Tmpl.nackJobSQL Text
schemaName Text
tableName (NominalDiffTime -> Int64
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling (NominalDiffTime -> Int64) -> Maybe NominalDiffTime -> Maybe Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe NominalDiffTime
delay) (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) (JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job) (JobRead payload -> Int32
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int32
attempts JobRead payload
job))
nackJobsBatch
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Maybe NominalDiffTime
-> [JobRead payload]
-> m [Int64]
nackJobsBatch :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Maybe NominalDiffTime -> [JobRead payload] -> m [Int64]
nackJobsBatch Text
_ Text
_ Maybe NominalDiffTime
_ [] = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
nackJobsBatch Text
schemaName Text
tableName Maybe NominalDiffTime
delay [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
-> Maybe Int64
-> [Int64]
-> [Int64]
-> [Int32]
-> Query Int64
Tmpl.nackJobsBatchSQL
Text
schemaName
Text
tableName
(NominalDiffTime -> Int64
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling (NominalDiffTime -> Int64) -> Maybe NominalDiffTime -> Maybe Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe NominalDiffTime
delay)
((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
Job 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.
Job 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.
Job payload Int64 q insertedAt adm -> Int32
attempts [JobRead payload]
jobs)
)
data TreeLocks
=
TakeLocks
|
LocksHeld TreesLocked
holdTreeLocks :: (Applicative m) => TreeLocks -> m TreesLocked -> m TreesLocked
holdTreeLocks :: forall (m :: * -> *).
Applicative m =>
TreeLocks -> m TreesLocked -> m TreesLocked
holdTreeLocks (LocksHeld TreesLocked
held) m TreesLocked
_ = TreesLocked -> m TreesLocked
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure TreesLocked
held
holdTreeLocks TreeLocks
TakeLocks m TreesLocked
takeLocks = m TreesLocked
takeLocks
moveToDLQ
:: forall m payload
. (MonadArbiter m)
=> TreeLocks
-> SchemaName
-> TableName
-> Text
-> 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 = 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 <- TreeLocks -> m TreesLocked -> m TreesLocked
forall (m :: * -> *).
Applicative m =>
TreeLocks -> m TreesLocked -> m TreesLocked
holdTreeLocks TreeLocks
locks (m TreesLocked -> m TreesLocked) -> m TreesLocked -> m TreesLocked
forall a b. (a -> b) -> a -> b
$ do
Text -> Text -> [(Int64, Maybe Int64)] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [(Int64, Maybe Int64)] -> m ()
lockJobsAndParents Text
schemaName Text
tableName [(Int64
jobId, JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe Int64
parentId JobRead payload
job)]
Text -> Text -> [Int64] -> m TreesLocked
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockJobTrees Text
schemaName Text
tableName [Int64
jobId]
rollup <- not . null <$> rollupIdsNow schemaName tableName [jobId]
moveToDLQFieldsInner
(LocksHeld held)
Tmpl.MoveNow
schemaName
tableName
errorMsg
jobId
(claimSeq job)
(parentId job)
rollup
where
jobId :: Int64
jobId = JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job
rollupIdsNow :: (MonadArbiter m) => SchemaName -> TableName -> [Int64] -> m [Int64]
rollupIdsNow :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m [Int64]
rollupIdsNow Text
schemaName Text
tableName [Int64]
ids = Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQueryPrepared (Text -> Text -> [Int64] -> Query Int64
Tmpl.rollupIdsSQL Text
schemaName Text
tableName [Int64]
ids)
moveToDLQFields
:: (MonadArbiter m)
=> TreeLocks
-> Tmpl.DLQMove
-> SchemaName
-> TableName
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> 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
$ 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
moveToDLQFieldsInner TreeLocks
locks DLQMove
move Text
schemaName Text
tableName Text
errorMsg Int64
jobId Int64
cseq Maybe Int64
mParentId Bool
rollup
moveToDLQFieldsInner
:: (MonadArbiter m)
=> TreeLocks
-> Tmpl.DLQMove
-> SchemaName
-> TableName
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> m Int64
moveToDLQFieldsInner :: forall (m :: * -> *).
MonadArbiter m =>
TreeLocks
-> DLQMove
-> Text
-> Text
-> Text
-> Int64
-> Int64
-> Maybe Int64
-> Bool
-> m Int64
moveToDLQFieldsInner TreeLocks
locks DLQMove
move Text
schemaName Text
tableName Text
errorMsg Int64
jobId Int64
cseq Maybe Int64
mParentId Bool
rollup = do
held <-
TreeLocks -> m TreesLocked -> m TreesLocked
forall (m :: * -> *).
Applicative m =>
TreeLocks -> m TreesLocked -> m TreesLocked
holdTreeLocks TreeLocks
locks (Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
lockJobParents Text
schemaName Text
tableName [Maybe Int64
mParentId] m () -> m TreesLocked -> m TreesLocked
forall a b. m a -> m b -> m b
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f b
*> Text -> Text -> [Int64] -> m TreesLocked
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockJobTrees Text
schemaName Text
tableName [Int64
jobId])
when rollup $ snapshotTreeRollups schemaName tableName jobId
rows <- countOr0 (Tmpl.moveToDLQSQL move schemaName tableName jobId cseq errorMsg)
when (rows > 0) $ do
when rollup $ void $ cascadeChildrenToDLQ schemaName tableName jobId parentMovedToDLQ
for_ mParentId $ \Int64
pid ->
TreesLocked -> Text -> Text -> Int64 -> m ()
forall (m :: * -> *).
MonadArbiter m =>
TreesLocked -> Text -> Text -> Int64 -> m ()
tryResumeParent TreesLocked
held Text
schemaName Text
tableName Int64
pid
pure rows
parentMovedToDLQ :: Text
parentMovedToDLQ :: Text
parentMovedToDLQ = Text
"Parent moved to DLQ"
cascadeChildrenToDLQ
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> Text
-> 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 :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne
Text
"cascadeChildrenToDLQ"
(Text -> Text -> Int64 -> Text -> Query Int64
Tmpl.cascadeChildrenToDLQSQL Text
schemaName Text
tableName Int64
parentJobId Text
errorMsg)
snapshotTreeRollups
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> 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)
moveToDLQBatch
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [(JobRead payload, Text)]
-> 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.
Job 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.
Job 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
Text -> Text -> [(Int64, Maybe Int64)] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [(Int64, Maybe Int64)] -> m ()
lockJobsAndParents Text
schemaName Text
tableName [(JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job, JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe Int64
parentId JobRead payload
job) | (JobRead payload
job, Text
_) <- [(JobRead payload, Text)]
jobsWithErrors]
rollupIds <- [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
<$> Text -> Text -> [Int64] -> m [Int64]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m [Int64]
rollupIdsNow Text
schemaName Text
tableName [Int64]
ids
held <- lockJobTrees schemaName tableName ids
for_ (Set.toAscList rollupIds) $ snapshotTreeRollups schemaName tableName
moved <- Set.fromList <$> MA.executeQuery (Tmpl.moveToDLQBatchSQL schemaName tableName ids cseqs 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.
Job 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 (Int64 -> Set Int64 -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) Set Int64
rollupIds)
(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.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job) Text
parentMovedToDLQ
resumeJobParents held schemaName tableName (map (parentId . fst) movedJobs)
pure (fromIntegral (Set.size moved))
retryFromDLQ
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Maybe (JobRead (Stored payload)))
retryFromDLQ :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
retryFromDLQ Text
schemaName Text
tableName Int64
dlqId = Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
retryFromDLQEditing Text
schemaName Text
tableName Int64
dlqId Maybe RowEdit
forall a. Maybe a
Nothing
retryFromDLQWithPayload
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
retryFromDLQWithPayload :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int64
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
retryFromDLQWithPayload Text
schemaName Text
tableName Int64
dlqId =
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
retryFromDLQEditing Text
schemaName Text
tableName Int64
dlqId (Maybe RowEdit -> m (Maybe (JobRead (Stored payload))))
-> (Maybe payload -> Maybe RowEdit)
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (payload -> RowEdit) -> Maybe payload -> Maybe RowEdit
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap payload -> RowEdit
forall payload. JobPayload payload => payload -> RowEdit
payloadEdit
retryFromDLQEditing
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
retryFromDLQEditing :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
retryFromDLQEditing Text
schemaName Text
tableName Int64
dlqId Maybe RowEdit
edit = m (Maybe (JobRead (Stored payload)))
-> m (Maybe (JobRead (Stored payload)))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Maybe (JobRead (Stored payload)))
-> m (Maybe (JobRead (Stored payload))))
-> m (Maybe (JobRead (Stored payload)))
-> m (Maybe (JobRead (Stored payload)))
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] -> m ())
-> ([Int64] -> [Maybe Int64]) -> [Int64] -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int64 -> Maybe Int64) -> [Int64] -> [Maybe Int64]
forall a b. (a -> b) -> [a] -> [b]
map Int64 -> Maybe Int64
forall a. a -> Maybe a
Just ([Int64] -> m ()) -> m [Int64] -> m ()
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m 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 -> Query Int64
Tmpl.dlqRetryParentSQL Text
schemaName Text
tableName Int64
dlqId)
m [Int64] -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (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.lockDLQRetryParentSQL Text
schemaName Text
tableName Int64
dlqId))
m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (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.lockDLQRetryGroupsSQL Text
schemaName Text
tableName Int64
dlqId))
Query (JobRead (Stored payload))
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text
-> Text
-> Int64
-> Maybe RowEdit
-> Query (JobRead (Stored payload))
forall payload.
Text
-> Text
-> Int64
-> Maybe RowEdit
-> Query (JobRead (Stored payload))
Tmpl.retryFromDLQSQL Text
schemaName Text
tableName Int64
dlqId Maybe RowEdit
edit)
retryFromDLQBatch
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> [Int64]
-> m Int64
retryFromDLQBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
retryFromDLQBatch Text
schemaName Text
tableName =
([Maybe (JobRead (Stored Value))] -> Int64)
-> m [Maybe (JobRead (Stored Value))] -> m Int64
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Int64)
-> ([Maybe (JobRead (Stored Value))] -> Int)
-> [Maybe (JobRead (Stored Value))]
-> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [JobRead (Stored Value)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length ([JobRead (Stored Value)] -> Int)
-> ([Maybe (JobRead (Stored Value))] -> [JobRead (Stored Value)])
-> [Maybe (JobRead (Stored Value))]
-> Int
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Maybe (JobRead (Stored Value))] -> [JobRead (Stored Value)]
forall a. [Maybe a] -> [a]
catMaybes) (m [Maybe (JobRead (Stored Value))] -> m Int64)
-> ([Int64] -> m [Maybe (JobRead (Stored Value))])
-> [Int64]
-> m Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int64 -> m (Maybe (JobRead (Stored Value))))
-> [Int64] -> m [Maybe (JobRead (Stored Value))]
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 (forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
retryFromDLQ @_ @Value Text
schemaName Text
tableName)
retryDLQFiltered
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> m Int64
retryDLQFiltered :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
retryDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters =
Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Query () -> Query Int64
Tmpl.dlqIdsFilteredSQL Text
schemaName Text
tableName (FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.DLQTable [JobFilter]
filters))
m [Int64] -> ([Int64] -> m Int64) -> m Int64
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
retryFromDLQBatch Text
schemaName Text
tableName
payloadEdit :: (JobPayload payload) => payload -> RowEdit
payloadEdit :: forall payload. JobPayload payload => payload -> RowEdit
payloadEdit = Codec (Value, PayloadColumns) PayloadKeys
-> (Value, PayloadColumns) -> RowEdit
forall s a. Codec s a -> s -> RowEdit
rowEdit Codec (Value, PayloadColumns) PayloadKeys
payloadWriteCodec ((Value, PayloadColumns) -> RowEdit)
-> (payload -> (Value, PayloadColumns)) -> payload -> RowEdit
forall b c a. (b -> c) -> (a -> b) -> a -> c
. payload -> (Value, PayloadColumns)
forall payload.
JobPayload payload =>
payload -> (Value, PayloadColumns)
payloadWrite
dlqJobExists
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m Bool
dlqJobExists :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Bool
dlqJobExists Text
schemaName Text
tableName Int64
dlqId =
Query Bool -> m Bool
forall (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> Text -> Int64 -> Query Bool
Tmpl.dlqJobExistsSQL Text
schemaName Text
tableName Int64
dlqId)
listDLQJobs
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int
-> Int
-> m [DLQ.DLQJob (Stored payload)]
listDLQJobs :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [DLQJob (Stored payload)]
listDLQJobs Text
schemaName Text
tableName = Text
-> Text -> [JobFilter] -> Int -> Int -> m [DLQJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> [JobFilter] -> Int -> Int -> m [DLQJob (Stored payload)]
listDLQFiltered Text
schemaName Text
tableName []
deleteDLQJob
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m Int64
deleteDLQJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
deleteDLQJob Text
schemaName Text
tableName Int64
dlqId = Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
deleteDLQJobsBatch Text
schemaName Text
tableName [Int64
dlqId]
deleteJobsResumingParents
:: (MonadArbiter m)
=> TreesLocked
-> SchemaName
-> TableName
-> Q.Query (Int64, Maybe Int64)
-> m [Int64]
deleteJobsResumingParents :: forall (m :: * -> *).
MonadArbiter m =>
TreesLocked
-> Text -> Text -> Query (Int64, Maybe Int64) -> m [Int64]
deleteJobsResumingParents TreesLocked
held Text
schemaName Text
tableName Query (Int64, Maybe Int64)
query = 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 Query (Int64, Maybe Int64)
query
resumeJobParents held schemaName tableName (map snd rows)
pure (map fst rows)
deleteDLQJobsBatch
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> [Int64]
-> m Int64
deleteDLQJobsBatch :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
deleteDLQJobsBatch Text
_ Text
_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
deleteDLQJobsBatch Text
schemaName Text
tableName [Int64]
dlqIds = 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
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.dlqParentIdsSQL Text
schemaName Text
tableName [Int64]
dlqIds)
lockJobParents schemaName tableName parents
held <- lockJobTrees schemaName tableName (catMaybes parents)
fromIntegral . length
<$> deleteJobsResumingParents held schemaName tableName (Tmpl.deleteDLQJobsBatchSQL schemaName tableName dlqIds)
purgeDLQ
:: (MonadArbiter m)
=> SchemaName
-> [TableName]
-> m (Int64, [Text])
purgeDLQ :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
purgeDLQ Text
schemaName [Text]
queues = do
retentions <- [(Text, Int64)] -> Map Text Int64
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(Text, Int64)] -> Map Text Int64)
-> m [(Text, Int64)] -> m (Map Text Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query (Text, Int64) -> m [(Text, Int64)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Query (Text, Int64)
Tmpl.queueDLQRetentionsSQL Text
schemaName)
let purgeOne Text
schema Text
queue = case Text -> Map Text Int64 -> Maybe Int64
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Text
queue Map Text Int64
retentions of
Maybe Int64
Nothing -> Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
Just Int64
retention ->
Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Text -> Text -> Int64 -> Int -> Query Int64
Tmpl.selectExpiredDLQSQL Text
schema Text
queue Int64
retention Int
dlqPurgeBatch)
m [Int64] -> ([Int64] -> m Int64) -> m Int64
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
deleteDLQJobsBatch Text
schema Text
queue
sweepQueues purgeOne schemaName (filter (`Map.member` retentions) queues)
dlqPurgeBatch :: Int
dlqPurgeBatch :: Int
dlqPurgeBatch = Int
1000
listArchiveFiltered
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> Maybe Tmpl.ArchiveSortColumn
-> Maybe Tmpl.SortDir
-> Int
-> Int
-> m [Archive.ArchiveJob (Stored payload)]
listArchiveFiltered :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob (Stored payload)]
listArchiveFiltered Text
schemaName Text
tableName [JobFilter]
filters Maybe ArchiveSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset =
((Int64, UTCTime, JobRead (Stored payload), Maybe Value)
-> ArchiveJob (Stored payload))
-> [(Int64, UTCTime, JobRead (Stored payload), Maybe Value)]
-> [ArchiveJob (Stored payload)]
forall a b. (a -> b) -> [a] -> [b]
map (Int64, UTCTime, JobRead (Stored payload), Maybe Value)
-> ArchiveJob (Stored payload)
forall payload.
(Int64, UTCTime, JobRead (Stored payload), Maybe Value)
-> ArchiveJob (Stored payload)
toArchiveRow
([(Int64, UTCTime, JobRead (Stored payload), Maybe Value)]
-> [ArchiveJob (Stored payload)])
-> m [(Int64, UTCTime, JobRead (Stored payload), Maybe Value)]
-> m [ArchiveJob (Stored payload)]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query (Int64, UTCTime, JobRead (Stored payload), Maybe Value)
-> m [(Int64, UTCTime, JobRead (Stored payload), Maybe Value)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
( Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (Int64, UTCTime, JobRead (Stored payload), Maybe Value)
forall payload.
Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (Int64, UTCTime, JobRead (Stored payload), Maybe Value)
Tmpl.listArchiveFilteredSQL
Text
schemaName
Text
tableName
(FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.ArchiveTable [JobFilter]
filters)
(Maybe ArchiveSortColumn -> Maybe SortDir -> Text
Tmpl.buildArchiveOrderBy Maybe ArchiveSortColumn
mSortBy Maybe SortDir
mSortDir)
(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)
)
listArchiveJobs
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int
-> Int
-> m [Archive.ArchiveJob (Stored payload)]
listArchiveJobs :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [ArchiveJob (Stored payload)]
listArchiveJobs Text
schemaName Text
tableName = Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob (Stored payload)]
listArchiveFiltered Text
schemaName Text
tableName [] Maybe ArchiveSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing
getArchiveJobById
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Maybe (Archive.ArchiveJob (Stored payload)))
getArchiveJobById :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (ArchiveJob (Stored payload)))
getArchiveJobById Text
schemaName Text
tableName Int64
jobId =
[ArchiveJob (Stored payload)]
-> Maybe (ArchiveJob (Stored payload))
forall a. [a] -> Maybe a
listToMaybe
([ArchiveJob (Stored payload)]
-> Maybe (ArchiveJob (Stored payload)))
-> m [ArchiveJob (Stored payload)]
-> m (Maybe (ArchiveJob (Stored payload)))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob (Stored 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
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 = Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
deleteArchiveJobsBatch Text
schemaName Text
tableName [Int64
archiveId]
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)
reEnqueueFromArchive
:: forall m payload
. (MonadArbiter m)
=> SchemaName -> TableName -> Int64 -> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchive :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchive Text
schemaName Text
tableName Int64
archiveId = Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchiveEditing Text
schemaName Text
tableName Int64
archiveId Maybe RowEdit
forall a. Maybe a
Nothing
reEnqueueFromArchiveWithPayload
:: forall m payload
. (JobPayload payload, MonadArbiter m)
=> SchemaName -> TableName -> Int64 -> Maybe payload -> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchiveWithPayload :: forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int64
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchiveWithPayload Text
schemaName Text
tableName Int64
archiveId =
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchiveEditing Text
schemaName Text
tableName Int64
archiveId (Maybe RowEdit -> m (Maybe (JobRead (Stored payload))))
-> (Maybe payload -> Maybe RowEdit)
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (payload -> RowEdit) -> Maybe payload -> Maybe RowEdit
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap payload -> RowEdit
forall payload. JobPayload payload => payload -> RowEdit
payloadEdit
reEnqueueFromArchiveEditing
:: (MonadArbiter m)
=> SchemaName -> TableName -> Int64 -> Maybe RowEdit -> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchiveEditing :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> Int64
-> Maybe RowEdit
-> m (Maybe (JobRead (Stored payload)))
reEnqueueFromArchiveEditing Text
schemaName Text
tableName Int64
archiveId Maybe RowEdit
edit =
m (Maybe (JobRead (Stored payload)))
-> m (Maybe (JobRead (Stored payload)))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Maybe (JobRead (Stored payload)))
-> m (Maybe (JobRead (Stored payload))))
-> m (Maybe (JobRead (Stored payload)))
-> m (Maybe (JobRead (Stored payload)))
forall a b. (a -> b) -> a -> b
$ Query (JobRead (Stored payload))
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text
-> Text
-> Int64
-> Maybe RowEdit
-> Query (JobRead (Stored payload))
forall payload.
Text
-> Text
-> Int64
-> Maybe RowEdit
-> Query (JobRead (Stored payload))
Tmpl.reEnqueueFromArchiveSQL Text
schemaName Text
tableName Int64
archiveId Maybe RowEdit
edit)
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)
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))
listArchiveJobsByGroup
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Text
-> Int
-> Int
-> m [Archive.ArchiveJob (Stored payload)]
listArchiveJobsByGroup :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Text -> Int -> Int -> m [ArchiveJob (Stored payload)]
listArchiveJobsByGroup Text
schemaName Text
tableName Text
key =
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [ArchiveJob (Stored 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
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 :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne
Text
"countArchiveFiltered"
(Text -> Text -> Query () -> Query Int64
Tmpl.countArchiveFilteredSQL Text
schemaName Text
tableName (FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.ArchiveTable [JobFilter]
filters))
toArchiveRow :: (Int64, UTCTime, JobRead (Stored payload), Maybe Value) -> Archive.ArchiveJob (Stored payload)
toArchiveRow :: forall payload.
(Int64, UTCTime, JobRead (Stored payload), Maybe Value)
-> ArchiveJob (Stored payload)
toArchiveRow (Int64
aId, UTCTime
aCompletedAt, JobRead (Stored payload)
snapshot, Maybe Value
aResult) =
Archive.ArchiveJob
{ archivePrimaryKey :: Int64
Archive.archivePrimaryKey = Int64
aId
, completedAt :: UTCTime
Archive.completedAt = UTCTime
aCompletedAt
, jobSnapshot :: JobRead (Stored payload)
Archive.jobSnapshot = JobRead (Stored payload)
snapshot
, archivedResult :: Maybe Value
Archive.archivedResult = Maybe Value
aResult
}
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 -> Text -> Int -> Query ()
Tmpl.purgeArchiveSQL Text
schemaName Text
queue Int
archivePurgeBatch))
archivePurgeBatch :: Int
archivePurgeBatch :: Int
archivePurgeBatch = Int
10000
listJobsFilteredOrdered
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> Maybe Tmpl.JobSortColumn
-> Maybe Tmpl.SortDir
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsFilteredOrdered :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [JobRead (Stored 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 (Stored payload), JobStatus) -> JobRead (Stored payload))
-> [(JobRead (Stored payload), JobStatus)]
-> [JobRead (Stored payload)]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead (Stored payload), JobStatus) -> JobRead (Stored payload)
forall a b. (a, b) -> a
fst ([(JobRead (Stored payload), JobStatus)]
-> [JobRead (Stored payload)])
-> m [(JobRead (Stored payload), JobStatus)]
-> m [JobRead (Stored payload)]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead (Stored payload), JobStatus)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead (Stored payload), JobStatus)]
listJobsWithStatus Text
schemaName Text
tableName [JobFilter]
filters Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset
| Bool
otherwise =
Query (JobRead (Stored payload)) -> m [JobRead (Stored payload)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Query (JobRead (Stored payload)) -> m [JobRead (Stored payload)])
-> Query (JobRead (Stored payload)) -> m [JobRead (Stored payload)]
forall a b. (a -> b) -> a -> b
$
Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (JobRead (Stored payload))
forall payload.
Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (JobRead (Stored payload))
Tmpl.listJobsFilteredSQL
Text
schemaName
Text
tableName
(FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.MainTable [JobFilter]
filters)
(Maybe JobSortColumn -> Maybe SortDir -> Text
Tmpl.buildJobsOrderBy Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir)
(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)
isStatusFilter :: Tmpl.JobFilter -> Bool
isStatusFilter :: JobFilter -> Bool
isStatusFilter (Tmpl.FilterStatus JobStatus
_) = Bool
True
isStatusFilter JobFilter
_ = Bool
False
jobRowWithStatusCodec :: TableName -> RowCodec (JobRead (Stored payload), Text)
jobRowWithStatusCodec :: forall payload. Text -> RowCodec (JobRead (Stored payload), Text)
jobRowWithStatusCodec Text
tableName =
(,) (JobRead (Stored payload)
-> Text -> (JobRead (Stored payload), Text))
-> Ap NullCol (JobRead (Stored payload))
-> Ap NullCol (Text -> (JobRead (Stored payload), Text))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Ap NullCol (JobRead (Stored payload))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName Ap NullCol (Text -> (JobRead (Stored payload), Text))
-> Ap NullCol Text -> Ap NullCol (JobRead (Stored payload), 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
listJobsWithStatus
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> Maybe Tmpl.JobSortColumn
-> Maybe Tmpl.SortDir
-> Int
-> Int
-> m [(JobRead (Stored payload), JobStatus)]
listJobsWithStatus :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead (Stored 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
(FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.MainTable [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 (Stored payload), Text)
-> m [(JobRead (Stored payload), Text)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (RowCodec (JobRead (Stored payload), Text)
-> Query () -> Query (JobRead (Stored payload), Text)
forall a. RowCodec a -> Query () -> Query a
Q.rows (Text -> RowCodec (JobRead (Stored payload), Text)
forall payload. Text -> RowCodec (JobRead (Stored payload), Text)
jobRowWithStatusCodec Text
tableName) Query ()
query)
traverse decodeJobStatusRow rows
decodeJobStatusRow
:: (MonadArbiter m)
=> (JobRead (Stored payload), Text)
-> m (JobRead (Stored payload), JobStatus)
decodeJobStatusRow :: forall (m :: * -> *) payload.
MonadArbiter m =>
(JobRead (Stored payload), Text)
-> m (JobRead (Stored payload), JobStatus)
decodeJobStatusRow (JobRead (Stored payload)
job, Text
rawStatus) = (,) JobRead (Stored payload)
job (JobStatus -> (JobRead (Stored payload), JobStatus))
-> m JobStatus -> m (JobRead (Stored payload), JobStatus)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Either Text JobStatus -> m JobStatus
forall (m :: * -> *) a. MonadIO m => Either Text a -> m a
orThrowParsing (Text -> Either Text JobStatus
jobStatusFromText Text
rawStatus)
listJobsFiltered
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsFiltered :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsFiltered Text
schemaName Text
tableName [JobFilter]
filters =
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [JobRead (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe JobSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing
countJobsFiltered
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> m Int64
countJobsFiltered :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
countJobsFiltered Text
schemaName Text
tableName [JobFilter]
filters =
Text -> Query Int64 -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne
Text
"countJobsFiltered"
(Text -> Text -> Query () -> Query Int64
Tmpl.countJobsFilteredSQL Text
schemaName Text
tableName (FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.MainTable [JobFilter]
filters))
listDLQFilteredOrdered
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> Maybe Tmpl.DLQSortColumn
-> Maybe Tmpl.SortDir
-> Int
-> Int
-> m [DLQ.DLQJob (Stored payload)]
listDLQFilteredOrdered :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob (Stored payload)]
listDLQFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe DLQSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset =
((Int64, UTCTime, JobRead (Stored payload))
-> DLQJob (Stored payload))
-> [(Int64, UTCTime, JobRead (Stored payload))]
-> [DLQJob (Stored payload)]
forall a b. (a -> b) -> [a] -> [b]
map (Int64, UTCTime, JobRead (Stored payload))
-> DLQJob (Stored payload)
forall payload.
(Int64, UTCTime, JobRead (Stored payload))
-> DLQJob (Stored payload)
toDLQRow
([(Int64, UTCTime, JobRead (Stored payload))]
-> [DLQJob (Stored payload)])
-> m [(Int64, UTCTime, JobRead (Stored payload))]
-> m [DLQJob (Stored payload)]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query (Int64, UTCTime, JobRead (Stored payload))
-> m [(Int64, UTCTime, JobRead (Stored payload))]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
( Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (Int64, UTCTime, JobRead (Stored payload))
forall payload.
Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (Int64, UTCTime, JobRead (Stored payload))
Tmpl.listDLQFilteredSQL
Text
schemaName
Text
tableName
(FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.DLQTable [JobFilter]
filters)
(Maybe DLQSortColumn -> Maybe SortDir -> Text
Tmpl.buildDLQOrderBy Maybe DLQSortColumn
mSortBy Maybe SortDir
mSortDir)
(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)
)
listDLQFiltered
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> Int
-> Int
-> m [DLQ.DLQJob (Stored payload)]
listDLQFiltered :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> [JobFilter] -> Int -> Int -> m [DLQJob (Stored payload)]
listDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters =
Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob (Stored payload)]
listDLQFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe DLQSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing
countDLQFiltered
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> [Tmpl.JobFilter]
-> 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 :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne
Text
"countDLQFiltered"
(Text -> Text -> Query () -> Query Int64
Tmpl.countDLQFilteredSQL Text
schemaName Text
tableName (FilterTable -> [JobFilter] -> Query ()
Tmpl.buildWhereClause FilterTable
Tmpl.DLQTable [JobFilter]
filters))
toDLQRow :: (Int64, UTCTime, JobRead (Stored payload)) -> DLQ.DLQJob (Stored payload)
toDLQRow :: forall payload.
(Int64, UTCTime, JobRead (Stored payload))
-> DLQJob (Stored payload)
toDLQRow (Int64
dlqId, UTCTime
dlqFailedAt, JobRead (Stored payload)
jobSnapshot) =
DLQ.DLQJob
{ dlqPrimaryKey :: Int64
DLQ.dlqPrimaryKey = Int64
dlqId
, failedAt :: UTCTime
DLQ.failedAt = UTCTime
dlqFailedAt
, jobSnapshot :: JobRead (Stored payload)
DLQ.jobSnapshot = JobRead (Stored payload)
jobSnapshot
}
listDLQJobsByParent
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> Int
-> Int
-> m [DLQ.DLQJob (Stored payload)]
listDLQJobsByParent :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> Int -> Int -> m [DLQJob (Stored payload)]
listDLQJobsByParent Text
schemaName Text
tableName Int64
parentJobId =
Text
-> Text -> [JobFilter] -> Int -> Int -> m [DLQJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> [JobFilter] -> Int -> Int -> m [DLQJob (Stored payload)]
listDLQFiltered Text
schemaName Text
tableName [Int64 -> JobFilter
Tmpl.FilterParentId Int64
parentJobId]
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]
listJobs
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobs :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [JobRead (Stored payload)]
listJobs Text
schemaName Text
tableName = Text
-> Text
-> [JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsFiltered Text
schemaName Text
tableName []
getJobById
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Maybe (JobRead (Stored payload)))
getJobById :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
getJobById Text
schemaName Text
tableName Int64
jobId =
Query (JobRead (Stored payload))
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Text -> Int64 -> Query (JobRead (Stored payload))
forall payload.
Text -> Text -> Int64 -> Query (JobRead (Stored payload))
Tmpl.getJobByIdSQL Text
schemaName Text
tableName Int64
jobId)
getJobByIdWithStatus
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m (Maybe (JobRead (Stored payload), JobStatus))
getJobByIdWithStatus :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Int64 -> m (Maybe (JobRead (Stored payload), JobStatus))
getJobByIdWithStatus Text
schemaName Text
tableName Int64
jobId = do
row <-
Query (JobRead (Stored payload), Text)
-> m (Maybe (JobRead (Stored payload), Text))
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (RowCodec (JobRead (Stored payload), Text)
-> Query () -> Query (JobRead (Stored payload), Text)
forall a. RowCodec a -> Query () -> Query a
Q.rows (Text -> RowCodec (JobRead (Stored payload), Text)
forall payload. Text -> RowCodec (JobRead (Stored payload), Text)
jobRowWithStatusCodec Text
tableName) (Text -> Text -> Int64 -> Query ()
Tmpl.getJobByIdWithStatusSQL Text
schemaName Text
tableName Int64
jobId))
traverse decodeJobStatusRow row
getJobByDedupKey
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Text
-> m (Maybe (JobRead (Stored payload)))
getJobByDedupKey :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Text -> m (Maybe (JobRead (Stored payload)))
getJobByDedupKey Text
schemaName Text
tableName Text
key =
Query (JobRead (Stored payload))
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Text -> Text -> Query (JobRead (Stored payload))
forall payload.
Text -> Text -> Text -> Query (JobRead (Stored payload))
Tmpl.getJobByDedupKeySQL Text
schemaName Text
tableName Text
key)
listJobsByGroup
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Text
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsByGroup :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Text -> Int -> Int -> m [JobRead (Stored payload)]
listJobsByGroup Text
schemaName Text
tableName Text
key =
Text
-> Text
-> [JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsFiltered Text
schemaName Text
tableName [Text -> JobFilter
Tmpl.FilterGroupKey Text
key]
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 =
Query Bool -> m Bool
forall (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> Text -> Int64 -> Query Bool
Tmpl.jobExistsSQL Text
schemaName Text
tableName Int64
jobId)
cancelJob
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> 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
$ 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)
lockParentAndSelf 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)
cancelJobsBatch
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> [Int64]
-> 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)
m TreesLocked -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m TreesLocked -> m ()) -> m TreesLocked -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> [Int64] -> m TreesLocked
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockParentsAndTrees Text
schemaName Text
tableName [Int64]
ids
[Int64] -> Int64
forall a. Num a => [a] -> a
forall (t :: * -> *) a. (Foldable t, Num a) => t a -> a
sum ([Int64] -> Int64) -> m [Int64] -> m Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (Int64 -> m Int64) -> [Int64] -> m [Int64]
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 (Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 (Query Int64 -> m Int64)
-> (Int64 -> Query Int64) -> Int64 -> m Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Text -> Int64 -> Query Int64
Tmpl.cancelJobSQL Text
schemaName Text
tableName) [Int64]
ids
promoteJob
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m Int64
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)
rescheduleJob
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> UTCTime
-> m Int64
rescheduleJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> UTCTime -> m Int64
rescheduleJob Text
schemaName Text
tableName Int64
jobId UTCTime
runAt =
Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
(Text -> Text -> Int64 -> UTCTime -> Query ()
Tmpl.rescheduleJobSQL Text
schemaName Text
tableName Int64
jobId UTCTime
runAt)
data QueueStats = QueueStats
{ QueueStats -> Int64
totalJobs :: Int64
, QueueStats -> Int64
readyJobs :: Int64
, QueueStats -> Int64
inFlightJobs :: Int64
, QueueStats -> Int64
scheduledJobs :: Int64
, QueueStats -> Int64
backoffJobs :: Int64
, QueueStats -> Int64
throttledJobs :: Int64
, QueueStats -> Int64
suspendedJobs :: Int64
, QueueStats -> Int64
cancelledJobs :: Int64
, QueueStats -> Int64
exhaustedJobs :: Int64
, QueueStats -> Int64
blockedJobs :: Int64
, QueueStats -> Maybe Double
oldestReadyAgeSeconds :: Maybe Double
, QueueStats -> Maybe Double
oldestInFlightAgeSeconds :: Maybe Double
, QueueStats -> Int64
dlqJobs :: Int64
, QueueStats -> Map Text Int64
kindCounts :: Map Text Int64
, QueueStats -> Map Text Int64
dlqKindCounts :: Map Text Int64
}
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
"exhaustedJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
exhaustedJobs QueueStats
stats
, Key
"blockedJobs" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= QueueStats -> Int64
blockedJobs 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
, Key
"dlqKindCounts" 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
dlqKindCounts 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats
QueueStats
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
(Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
"exhaustedJobs"
Parser
(Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser Int64
-> Parser
(Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
"blockedJobs"
Parser
(Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Parser (Maybe Double)
-> Parser
(Maybe Double
-> Int64 -> Map Text 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 -> Map Text Int64 -> QueueStats)
-> Parser (Maybe Double)
-> Parser (Int64 -> Map Text 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 -> Map Text Int64 -> QueueStats)
-> Parser Int64
-> Parser (Map Text 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
"dlqJobs"
Parser (Map Text Int64 -> Map Text Int64 -> QueueStats)
-> Parser (Map Text 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 (Map Text Int64)
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"kindCounts"
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 (Map Text Int64)
forall a. FromJSON a => Object -> Key -> Parser a
.: Key
"dlqKindCounts"
emptyQueueStats :: QueueStats
emptyQueueStats :: QueueStats
emptyQueueStats =
QueueStats
{ totalJobs :: Int64
totalJobs = Int64
0
, readyJobs :: Int64
readyJobs = Int64
0
, inFlightJobs :: Int64
inFlightJobs = Int64
0
, scheduledJobs :: Int64
scheduledJobs = Int64
0
, backoffJobs :: Int64
backoffJobs = Int64
0
, throttledJobs :: Int64
throttledJobs = Int64
0
, suspendedJobs :: Int64
suspendedJobs = Int64
0
, cancelledJobs :: Int64
cancelledJobs = Int64
0
, exhaustedJobs :: Int64
exhaustedJobs = Int64
0
, blockedJobs :: Int64
blockedJobs = Int64
0
, oldestReadyAgeSeconds :: Maybe Double
oldestReadyAgeSeconds = Maybe Double
forall a. Maybe a
Nothing
, oldestInFlightAgeSeconds :: Maybe Double
oldestInFlightAgeSeconds = Maybe Double
forall a. Maybe a
Nothing
, dlqJobs :: Int64
dlqJobs = Int64
0
, kindCounts :: Map Text Int64
kindCounts = Map Text Int64
forall k a. Map k a
Map.empty
, dlqKindCounts :: Map Text Int64
dlqKindCounts = Map Text Int64
forall k a. Map k a
Map.empty
}
queueStatusCounts :: QueueStats -> [(Text, Int64)]
queueStatusCounts :: QueueStats -> [(Text, Int64)]
queueStatusCounts QueueStats
stats =
[ (JobStatus -> Text
jobStatusToText JobStatus
Ready, QueueStats -> Int64
readyJobs QueueStats
stats)
, (Text
blockedStatusLabel, QueueStats -> Int64
blockedJobs QueueStats
stats)
, (JobStatus -> Text
jobStatusToText JobStatus
InFlight, QueueStats -> Int64
inFlightJobs QueueStats
stats)
, (JobStatus -> Text
jobStatusToText JobStatus
Scheduled, QueueStats -> Int64
scheduledJobs QueueStats
stats)
, (JobStatus -> Text
jobStatusToText JobStatus
Backoff, QueueStats -> Int64
backoffJobs QueueStats
stats)
, (JobStatus -> Text
jobStatusToText JobStatus
Throttled, QueueStats -> Int64
throttledJobs QueueStats
stats)
, (JobStatus -> Text
jobStatusToText JobStatus
Suspended, QueueStats -> Int64
suspendedJobs QueueStats
stats)
, (JobStatus -> Text
jobStatusToText JobStatus
Cancelled, QueueStats -> Int64
cancelledJobs QueueStats
stats)
, (JobStatus -> Text
jobStatusToText JobStatus
Exhausted, QueueStats -> Int64
exhaustedJobs QueueStats
stats)
]
blockedStatusLabel :: Text
blockedStatusLabel :: Text
blockedStatusLabel = Text
"blocked"
statsRowCodec :: RowCodec QueueStats
statsRowCodec :: RowCodec QueueStats
statsRowCodec =
Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats
QueueStats
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
-> Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
(Int64
-> Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
"exhausted_jobs" Col Int64
CInt8
Ap
NullCol
(Int64
-> Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol Int64
-> Ap
NullCol
(Maybe Double
-> Maybe Double
-> Int64
-> Map Text 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
"blocked_jobs" Col Int64
CInt8
Ap
NullCol
(Maybe Double
-> Maybe Double
-> Int64
-> Map Text Int64
-> Map Text Int64
-> QueueStats)
-> Ap NullCol (Maybe Double)
-> Ap
NullCol
(Maybe Double
-> Int64 -> Map Text 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 -> Map Text Int64 -> QueueStats)
-> Ap NullCol (Maybe Double)
-> Ap
NullCol (Int64 -> Map Text 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 -> Map Text Int64 -> QueueStats)
-> Ap NullCol Int64
-> Ap NullCol (Map Text 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
"dlq_jobs" Col Int64
CInt8
Ap NullCol (Map Text Int64 -> Map Text Int64 -> QueueStats)
-> Ap NullCol (Map Text 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
<*> (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)
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
"dlq_kind_counts" Col Value
CJsonb)
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 Value -> Map Text Int64
forall a. (FromJSON a, Monoid a) => Value -> a
decodeOrEmpty
getQueueStats
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> [Text]
-> m QueueStats
getQueueStats :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Text] -> m QueueStats
getQueueStats Text
schemaName Text
tableName [Text]
kinds =
QueueStats -> Maybe QueueStats -> QueueStats
forall a. a -> Maybe a -> a
fromMaybe QueueStats
emptyQueueStats (Maybe QueueStats -> QueueStats)
-> m (Maybe QueueStats) -> m QueueStats
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query QueueStats -> m (Maybe QueueStats)
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (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)
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
getAllQueueStats
:: (MonadArbiter m)
=> SchemaName
-> [(TableName, [Text])]
-> 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)
data GroupSummary = GroupSummary
{ GroupSummary -> Text
gsGroupKey :: Text
, GroupSummary -> Int64
gsJobCount :: Int64
, GroupSummary -> Int64
gsReadyCount :: Int64
, GroupSummary -> Maybe UTCTime
gsNextDue :: Maybe UTCTime
, GroupSummary -> Maybe UTCTime
gsInFlightUntil :: Maybe UTCTime
, GroupSummary -> Bool
gsInFlight :: Bool
, GroupSummary -> Maybe Int64
gsHeadJobId :: Maybe Int64
, GroupSummary -> Maybe JobStatus
gsHeadStatus :: Maybe JobStatus
, GroupSummary -> Bool
gsHeadBlocked :: Bool
}
deriving stock (GroupSummary -> GroupSummary -> Bool
(GroupSummary -> GroupSummary -> Bool)
-> (GroupSummary -> GroupSummary -> Bool) -> Eq GroupSummary
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: GroupSummary -> GroupSummary -> Bool
== :: GroupSummary -> GroupSummary -> Bool
$c/= :: GroupSummary -> GroupSummary -> Bool
/= :: GroupSummary -> GroupSummary -> Bool
Eq, Int -> GroupSummary -> ShowS
[GroupSummary] -> ShowS
GroupSummary -> String
(Int -> GroupSummary -> ShowS)
-> (GroupSummary -> String)
-> ([GroupSummary] -> ShowS)
-> Show GroupSummary
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> GroupSummary -> ShowS
showsPrec :: Int -> GroupSummary -> ShowS
$cshow :: GroupSummary -> String
show :: GroupSummary -> String
$cshowList :: [GroupSummary] -> ShowS
showList :: [GroupSummary] -> ShowS
Show)
instance ToJSON GroupSummary where
toJSON :: GroupSummary -> Value
toJSON GroupSummary
summary =
[Pair] -> Value
object
[ Key
"groupKey" Key -> Text -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Text
gsGroupKey GroupSummary
summary
, Key
"jobCount" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Int64
gsJobCount GroupSummary
summary
, Key
"readyCount" Key -> Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Int64
gsReadyCount GroupSummary
summary
, Key
"nextDue" Key -> Maybe UTCTime -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Maybe UTCTime
gsNextDue GroupSummary
summary
, Key
"inFlightUntil" Key -> Maybe UTCTime -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Maybe UTCTime
gsInFlightUntil GroupSummary
summary
, Key
"inFlight" Key -> Bool -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Bool
gsInFlight GroupSummary
summary
, Key
"headJobId" Key -> Maybe Int64 -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Maybe Int64
gsHeadJobId GroupSummary
summary
, Key
"headStatus" Key -> Maybe JobStatus -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Maybe JobStatus
gsHeadStatus GroupSummary
summary
, Key
"headBlocked" Key -> Bool -> Pair
forall v. ToJSON v => Key -> v -> Pair
forall e kv v. (KeyValue e kv, ToJSON v) => Key -> v -> kv
.= GroupSummary -> Bool
gsHeadBlocked GroupSummary
summary
]
instance FromJSON GroupSummary where
parseJSON :: Value -> Parser GroupSummary
parseJSON = String
-> (Object -> Parser GroupSummary) -> Value -> Parser GroupSummary
forall a. String -> (Object -> Parser a) -> Value -> Parser a
withObject String
"GroupSummary" ((Object -> Parser GroupSummary) -> Value -> Parser GroupSummary)
-> (Object -> Parser GroupSummary) -> Value -> Parser GroupSummary
forall a b. (a -> b) -> a -> b
$ \Object
obj ->
Text
-> Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary
GroupSummary
(Text
-> Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Parser Text
-> Parser
(Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
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
"groupKey"
Parser
(Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Parser Int64
-> Parser
(Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
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
"jobCount"
Parser
(Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Parser Int64
-> Parser
(Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
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
"readyCount"
Parser
(Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Parser (Maybe UTCTime)
-> Parser
(Maybe UTCTime
-> Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
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 UTCTime)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"nextDue"
Parser
(Maybe UTCTime
-> Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
-> Parser (Maybe UTCTime)
-> Parser
(Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
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 UTCTime)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"inFlightUntil"
Parser
(Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
-> Parser Bool
-> Parser (Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
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
"inFlight"
Parser (Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
-> Parser (Maybe Int64)
-> Parser (Maybe JobStatus -> Bool -> GroupSummary)
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 Int64)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"headJobId"
Parser (Maybe JobStatus -> Bool -> GroupSummary)
-> Parser (Maybe JobStatus) -> Parser (Bool -> GroupSummary)
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 JobStatus)
forall a. FromJSON a => Object -> Key -> Parser (Maybe a)
.:? Key
"headStatus"
Parser (Bool -> GroupSummary) -> Parser Bool -> Parser GroupSummary
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
"headBlocked"
groupSummaryCodec :: RowCodec (GroupSummary, Maybe Text)
groupSummaryCodec :: RowCodec (GroupSummary, Maybe Text)
groupSummaryCodec =
(,)
(GroupSummary -> Maybe Text -> (GroupSummary, Maybe Text))
-> Ap NullCol GroupSummary
-> Ap NullCol (Maybe Text -> (GroupSummary, Maybe Text))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ( Text
-> Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary
GroupSummary
(Text
-> Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Ap NullCol Text
-> Ap
NullCol
(Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
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
"group_key" Col Text
CText
Ap
NullCol
(Int64
-> Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Ap NullCol Int64
-> Ap
NullCol
(Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
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
"job_count" Col Int64
CInt8
Ap
NullCol
(Int64
-> Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Ap NullCol Int64
-> Ap
NullCol
(Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
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_count" Col Int64
CInt8
Ap
NullCol
(Maybe UTCTime
-> Maybe UTCTime
-> Bool
-> Maybe Int64
-> Maybe JobStatus
-> Bool
-> GroupSummary)
-> Ap NullCol (Maybe UTCTime)
-> Ap
NullCol
(Maybe UTCTime
-> Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
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 UTCTime -> Ap NullCol (Maybe UTCTime)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"next_due" Col UTCTime
CTimestamptz
Ap
NullCol
(Maybe UTCTime
-> Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
-> Ap NullCol (Maybe UTCTime)
-> Ap
NullCol
(Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
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 UTCTime -> Ap NullCol (Maybe UTCTime)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"in_flight_until" Col UTCTime
CTimestamptz
Ap
NullCol
(Bool -> Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
-> Ap NullCol Bool
-> Ap
NullCol (Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
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
"in_flight" Col Bool
CBool
Ap NullCol (Maybe Int64 -> Maybe JobStatus -> Bool -> GroupSummary)
-> Ap NullCol (Maybe Int64)
-> Ap NullCol (Maybe JobStatus -> Bool -> GroupSummary)
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
"head_id" Col Int64
CInt8
Ap NullCol (Maybe JobStatus -> Bool -> GroupSummary)
-> Ap NullCol (Maybe JobStatus)
-> Ap NullCol (Bool -> GroupSummary)
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 JobStatus -> Ap NullCol (Maybe JobStatus)
forall a. a -> Ap NullCol a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe JobStatus
forall a. Maybe a
Nothing
Ap NullCol (Bool -> GroupSummary)
-> Ap NullCol Bool -> Ap NullCol GroupSummary
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
"head_blocked" Col Bool
CBool
)
Ap NullCol (Maybe Text -> (GroupSummary, Maybe Text))
-> Ap NullCol (Maybe Text) -> RowCodec (GroupSummary, Maybe 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 (Maybe Text)
forall a. Text -> Col a -> RowCodec (Maybe a)
ncol Text
"head_status" Col Text
CText
listGroups
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Maybe Text
-> Int
-> Int
-> m [GroupSummary]
listGroups :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe Text -> Int -> Int -> m [GroupSummary]
listGroups Text
schemaName Text
tableName Maybe Text
key Int
limit Int
offset = do
found <-
Query (GroupSummary, Maybe Text) -> m [(GroupSummary, Maybe Text)]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery
(RowCodec (GroupSummary, Maybe Text)
-> Text
-> Text
-> Maybe Text
-> Int64
-> Int64
-> Query (GroupSummary, Maybe Text)
forall a.
RowCodec a
-> Text -> Text -> Maybe Text -> Int64 -> Int64 -> Query a
Tmpl.listGroupsSQL RowCodec (GroupSummary, Maybe Text)
groupSummaryCodec Text
schemaName Text
tableName Maybe Text
key (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 withStatus found
where
withStatus :: (GroupSummary, Maybe Text) -> m GroupSummary
withStatus (GroupSummary
summary, Maybe Text
rawStatus) = do
status <- (Text -> m JobStatus) -> Maybe Text -> m (Maybe JobStatus)
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) -> Maybe a -> f (Maybe b)
traverse (Either Text JobStatus -> m JobStatus
forall (m :: * -> *) a. MonadIO m => Either Text a -> m a
orThrowParsing (Either Text JobStatus -> m JobStatus)
-> (Text -> Either Text JobStatus) -> Text -> m JobStatus
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Either Text JobStatus
jobStatusFromText) Maybe Text
rawStatus
pure summary {gsHeadStatus = status}
countGroups :: (MonadArbiter m) => SchemaName -> TableName -> Maybe Text -> m Int64
countGroups :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe Text -> m Int64
countGroups Text
schemaName Text
tableName Maybe Text
key =
Text -> Query Int64 -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne Text
"countGroups" (Text -> Text -> Maybe Text -> Query Int64
Tmpl.countGroupsSQL Text
schemaName Text
tableName Maybe Text
key)
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 []
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]
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 []
listJobsByParent
:: forall m payload
. (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsByParent :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> Int -> Int -> m [JobRead (Stored payload)]
listJobsByParent Text
schemaName Text
tableName Int64
pid =
Text
-> Text
-> [JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Int
-> Int
-> m [JobRead (Stored payload)]
listJobsFiltered Text
schemaName Text
tableName [Int64 -> JobFilter
Tmpl.FilterParentId Int64
pid]
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]
countChildrenBatch
:: (MonadArbiter m)
=> SchemaName -> TableName -> [Int64] -> m (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 =
[(Int64, (Int64, Int64))] -> Map Int64 (Int64, Int64)
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(Int64, (Int64, Int64))] -> Map Int64 (Int64, Int64))
-> m [(Int64, (Int64, Int64))] -> m (Map Int64 (Int64, Int64))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> 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 (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))
countDLQChildrenBatch
:: (MonadArbiter m)
=> SchemaName -> TableName -> [Int64] -> m (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 =
[(Int64, Int64)] -> Map Int64 Int64
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(Int64, Int64)] -> Map Int64 Int64)
-> m [(Int64, Int64)] -> m (Map Int64 Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> 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)
pauseChildren
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> 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)
resumeChildren
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> 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)
cancelJobCascade
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m [Int64]
cancelJobCascade :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m [Int64]
cancelJobCascade Text
schemaName Text
tableName =
([Int64] -> Bool)
-> (Int64 -> m [Int64]) -> Text -> Text -> Int64 -> m [Int64]
forall (m :: * -> *) a.
MonadArbiter m =>
(a -> Bool) -> (Int64 -> m a) -> Text -> Text -> Int64 -> m a
cascadeDeleteJob (Bool -> Bool
not (Bool -> Bool) -> ([Int64] -> Bool) -> [Int64] -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int64] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null) (Query Int64 -> m [Int64]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery (Query Int64 -> m [Int64])
-> (Int64 -> Query Int64) -> Int64 -> m [Int64]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Text -> Int64 -> Query Int64
Tmpl.cancelJobCascadeSQL Text
schemaName Text
tableName) Text
schemaName Text
tableName
cascadeDeleteJob
:: (MonadArbiter m)
=> (a -> Bool)
-> (Int64 -> m a)
-> SchemaName
-> TableName
-> Int64
-> m a
cascadeDeleteJob :: forall (m :: * -> *) a.
MonadArbiter m =>
(a -> Bool) -> (Int64 -> m a) -> Text -> Text -> Int64 -> m a
cascadeDeleteJob a -> Bool
deletedAny Int64 -> m a
delete Text
schemaName Text
tableName Int64
jobId = m a -> m a
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m a -> m a) -> m a -> m a
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)
lockParentAndSelf Text
schemaName Text
tableName Int64
jobId
held <- lockJobTrees schemaName tableName [jobId]
deleted <- delete jobId
when (deletedAny deleted)
$ for_ rootParentId
$ tryResumeParent held schemaName tableName
pure deleted
cancelJobTree
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m [Int64]
cancelJobTree :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m [Int64]
cancelJobTree Text
schemaName Text
tableName Int64
jobId =
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.cancelJobTreeSQL Text
schemaName Text
tableName Int64
jobId)
forceCancelJob
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> m Int64
forceCancelJob :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
forceCancelJob Text
schemaName Text
tableName =
(Int64 -> Bool)
-> (Int64 -> m Int64) -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *) a.
MonadArbiter m =>
(a -> Bool) -> (Int64 -> m a) -> Text -> Text -> Int64 -> m a
cascadeDeleteJob (Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
> Int64
0) (Query Int64 -> m Int64
forall (m :: * -> *). MonadArbiter m => Query Int64 -> m Int64
countOr0 (Query Int64 -> m Int64)
-> (Int64 -> Query Int64) -> Int64 -> m Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Text -> Int64 -> Query Int64
Tmpl.forceCancelJobSQL Text
schemaName Text
tableName) Text
schemaName Text
tableName
data CancelledScope payload
=
LapsedOnly [Int64]
|
HeldBy UUID [JobRead payload]
deleteCancelledJobs
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> CancelledScope payload
-> m [Int64]
deleteCancelledJobs :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> CancelledScope payload -> m [Int64]
deleteCancelledJobs Text
schemaName Text
tableName CancelledScope payload
scope
| [Int64] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Int64]
jobIds = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
| Bool
otherwise = 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 <- Text -> Text -> [Int64] -> m TreesLocked
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
lockParentsAndTrees Text
schemaName Text
tableName [Int64]
jobIds
deleteJobsResumingParents held schemaName tableName deleteSql
where
([Int64]
jobIds, Query (Int64, Maybe Int64)
deleteSql) = case CancelledScope payload
scope of
LapsedOnly [Int64]
ids -> ([Int64]
ids, Text -> Text -> [Int64] -> Query (Int64, Maybe Int64)
Tmpl.deleteLapsedCancelledJobsSQL Text
schemaName Text
tableName [Int64]
ids)
HeldBy UUID
owner [JobRead payload]
jobs ->
( (JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey [JobRead payload]
jobs
, Text
-> Text -> UUID -> [(Int64, Int64)] -> Query (Int64, Maybe Int64)
Tmpl.deleteHeldCancelledJobsSQL Text
schemaName Text
tableName UUID
owner [(JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job, JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq JobRead payload
job) | JobRead payload
job <- [JobRead payload]
jobs]
)
suspendJob
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> 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)
resumeJob
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int64
-> 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)
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 {groupsWindowFrom :: GroupsCursor -> Maybe Text
groupsWindowFrom = Maybe Text
window, groupsEmptiedFrom :: GroupsCursor -> Maybe Text
groupsEmptiedFrom = 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"
data GroupsPass = GroupsPass
{ GroupsPass -> Int64
passRewritten :: Int64
, GroupsPass -> Bool
passRepairFailed :: Bool
, GroupsPass -> Maybe GroupsCursor
passResume :: Maybe GroupsCursor
}
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 GroupsCursor {groupsWindowFrom :: Maybe Text
groupsWindowFrom = Maybe Text
window, groupsEmptiedFrom :: Maybe Text
groupsEmptiedFrom = Maybe Text
emptied}
refreshGroupsForQueue
:: (MonadArbiter m)
=> SchemaName
-> TableName
-> Int
-> Maybe GroupsCursor
-> 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)
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
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
lastRow :: [a] -> Maybe a
lastRow :: forall a. [a] -> Maybe a
lastRow = (([a], a) -> a) -> Maybe ([a], a) -> Maybe a
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ([a], a) -> a
forall a b. (a, b) -> b
snd (Maybe ([a], a) -> Maybe a)
-> ([a] -> Maybe ([a], a)) -> [a] -> Maybe a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [a] -> Maybe ([a], a)
forall a. [a] -> Maybe ([a], a)
unsnoc
groupsRefreshBatch :: Int
groupsRefreshBatch :: Int
groupsRefreshBatch = Int
20000
refreshAllGroups
:: (MonadArbiter m)
=> SchemaName
-> [TableName]
-> Map TableName GroupsCursor
-> m ((Int64, [Text]), 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)
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)
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)
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
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
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]
held <- lockJobTrees schemaName tableName ids
sum <$> traverse (moveOne held) exhausted
where
moveOne :: TreesLocked -> (Int64, Int64, Maybe Int64, Bool) -> m Int64
moveOne TreesLocked
held (Int64
jobId, Int64
cseq, Maybe Int64
mParentId, Bool
rollup) =
Int64 -> Either SomeException Int64 -> Int64
forall b a. b -> Either a b -> b
fromRight Int64
0
(Either SomeException Int64 -> Int64)
-> m (Either SomeException Int64) -> m 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 (TreesLocked -> TreeLocks
LocksHeld TreesLocked
held) 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)"
exhaustedSweepBatch :: Int
exhaustedSweepBatch :: Int
exhaustedSweepBatch = Int
1000
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
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 (LapsedOnly ids)
cancelledSweepBatch :: Int
cancelledSweepBatch :: Int
cancelledSweepBatch = Int
1000
upsertCronDefault
:: (MonadArbiter m)
=> SchemaName
-> Text
-> Text
-> Text
-> Text
-> Maybe Text
-> Bool
-> m Int64
upsertCronDefault :: forall (m :: * -> *).
MonadArbiter m =>
Text
-> Text -> Text -> Text -> Text -> Maybe Text -> Bool -> m Int64
upsertCronDefault Text
schemaName Text
scheduleName Text
queueName Text
defaultExpr Text
defaultOv Maybe Text
defaultTz Bool
initialEnabled =
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 -> Bool -> Query ()
Tmpl.upsertCronDefaultSQL Text
schemaName Text
scheduleName Text
queueName Text
defaultExpr Text
defaultOv Maybe Text
defaultTz Bool
initialEnabled)
listCronSchedules
:: (MonadArbiter m)
=> SchemaName
-> Maybe Text
-> 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)
getCronScheduleByName
:: (MonadArbiter m)
=> SchemaName
-> Text
-> m (Maybe CronScheduleRow)
getCronScheduleByName :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
getCronScheduleByName Text
schemaName Text
scheduleName =
Query CronScheduleRow -> m (Maybe CronScheduleRow)
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Text -> Query CronScheduleRow
Tmpl.getCronScheduleByNameSQL Text
schemaName Text
scheduleName)
updateCronSchedule
:: (MonadArbiter m)
=> SchemaName
-> Text
-> 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)
touchCronChecked
:: (MonadArbiter m)
=> SchemaName
-> [Text]
-> UTCTime
-> m Int64
touchCronChecked :: forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> UTCTime -> 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 [Text]
names UTCTime
watermark =
Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
(Text -> [Text] -> UTCTime -> Query ()
Tmpl.touchCronCheckedSQL Text
schemaName [Text]
names UTCTime
watermark)
tryFireCronGate
:: (MonadArbiter m)
=> SchemaName
-> Text
-> UTCTime
-> m Bool
tryFireCronGate :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> UTCTime -> m Bool
tryFireCronGate Text
schemaName Text
scheduleName UTCTime
minuteFloor =
(Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
> Int64
0) (Int64 -> Bool) -> m Int64 -> m Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Text -> UTCTime -> Query ()
Tmpl.tryFireCronGateSQL Text
schemaName Text
scheduleName UTCTime
minuteFloor)
tryAcquireCronLeader
:: (MonadArbiter m)
=> SchemaName
-> Text
-> Text
-> m Bool
tryAcquireCronLeader :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> m Bool
tryAcquireCronLeader Text
schemaName Text
queueName Text
scheduleName =
Query Bool -> m Bool
forall (m :: * -> *). MonadArbiter m => Query Bool -> m Bool
MA.queryAny (Text -> Text -> Text -> Query Bool
Tmpl.tryAcquireCronLeaderSQL Text
schemaName Text
queueName Text
scheduleName)
data RunRequestOutcome
=
RunReqNotFound
|
RunReqDisabled
|
RunReqStamped
|
RunReqPending
deriving stock (RunRequestOutcome
RunRequestOutcome -> RunRequestOutcome -> Bounded RunRequestOutcome
forall a. a -> a -> Bounded a
$cminBound :: RunRequestOutcome
minBound :: RunRequestOutcome
$cmaxBound :: RunRequestOutcome
maxBound :: RunRequestOutcome
Bounded, Int -> RunRequestOutcome
RunRequestOutcome -> Int
RunRequestOutcome -> [RunRequestOutcome]
RunRequestOutcome -> RunRequestOutcome
RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
RunRequestOutcome
-> RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
(RunRequestOutcome -> RunRequestOutcome)
-> (RunRequestOutcome -> RunRequestOutcome)
-> (Int -> RunRequestOutcome)
-> (RunRequestOutcome -> Int)
-> (RunRequestOutcome -> [RunRequestOutcome])
-> (RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome])
-> (RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome])
-> (RunRequestOutcome
-> RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome])
-> Enum RunRequestOutcome
forall a.
(a -> a)
-> (a -> a)
-> (Int -> a)
-> (a -> Int)
-> (a -> [a])
-> (a -> a -> [a])
-> (a -> a -> [a])
-> (a -> a -> a -> [a])
-> Enum a
$csucc :: RunRequestOutcome -> RunRequestOutcome
succ :: RunRequestOutcome -> RunRequestOutcome
$cpred :: RunRequestOutcome -> RunRequestOutcome
pred :: RunRequestOutcome -> RunRequestOutcome
$ctoEnum :: Int -> RunRequestOutcome
toEnum :: Int -> RunRequestOutcome
$cfromEnum :: RunRequestOutcome -> Int
fromEnum :: RunRequestOutcome -> Int
$cenumFrom :: RunRequestOutcome -> [RunRequestOutcome]
enumFrom :: RunRequestOutcome -> [RunRequestOutcome]
$cenumFromThen :: RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
enumFromThen :: RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
$cenumFromTo :: RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
enumFromTo :: RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
$cenumFromThenTo :: RunRequestOutcome
-> RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
enumFromThenTo :: RunRequestOutcome
-> RunRequestOutcome -> RunRequestOutcome -> [RunRequestOutcome]
Enum, 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)
runRequestOutcomeToText :: RunRequestOutcome -> Text
runRequestOutcomeToText :: RunRequestOutcome -> Text
runRequestOutcomeToText = \case
RunRequestOutcome
RunReqNotFound -> Text
"not_found"
RunRequestOutcome
RunReqDisabled -> Text
"disabled"
RunRequestOutcome
RunReqStamped -> Text
"stamped"
RunRequestOutcome
RunReqPending -> Text
"pending"
requestCronRun
:: (MonadArbiter m)
=> SchemaName
-> Text
-> m RunRequestOutcome
requestCronRun :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
requestCronRun Text
schemaName Text
scheduleName = do
status <- Text -> Query Text -> m Text
forall (m :: * -> *) a. MonadArbiter m => Text -> Query a -> m a
queryOne Text
"requestCronRun" (Text -> Text -> Query Text
Tmpl.requestCronRunSQL Text
schemaName Text
scheduleName)
orThrowParsing . first ("requestCronRun: " <>) $
enumFromText "run request outcome" runRequestOutcomeToText status
claimCronRun
:: (MonadArbiter m)
=> SchemaName
-> Text
-> m (Maybe CronScheduleRow)
claimCronRun :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
claimCronRun Text
schemaName Text
scheduleName =
Query CronScheduleRow -> m (Maybe CronScheduleRow)
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Text -> Query CronScheduleRow
Tmpl.claimCronRunSQL Text
schemaName Text
scheduleName)
touchCronManualRun
:: (MonadArbiter m)
=> SchemaName
-> Text
-> UTCTime
-> m Int64
touchCronManualRun :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> UTCTime -> m Int64
touchCronManualRun Text
schemaName Text
scheduleName UTCTime
firedAt =
Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement
(Text -> Text -> UTCTime -> Query ()
Tmpl.touchCronManualRunSQL Text
schemaName Text
scheduleName UTCTime
firedAt)
pendingCronRuns
:: (MonadArbiter m)
=> SchemaName
-> [Text]
-> 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)
ensureQueue
:: (MonadArbiter m)
=> SchemaName
-> Text
-> 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)
setQueuePaused
:: (MonadArbiter m)
=> SchemaName
-> Text
-> 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)
setQueueDLQRetention
:: (MonadArbiter m)
=> SchemaName
-> Text
-> Maybe NominalDiffTime
-> m ()
setQueueDLQRetention :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe NominalDiffTime -> m ()
setQueueDLQRetention Text
schemaName Text
queue Maybe NominalDiffTime
retention =
m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement (Text -> Text -> Maybe Int64 -> Query ()
Tmpl.setQueueDLQRetentionSQL Text
schemaName Text
queue (NominalDiffTime -> Int64
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling (NominalDiffTime -> Int64) -> Maybe NominalDiffTime -> Maybe Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe NominalDiffTime
retention)))
schemaQueues
:: (MonadArbiter m)
=> SchemaName
-> [TableName]
-> m [TableName]
schemaQueues :: forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m [Text]
schemaQueues Text
schemaName [Text]
queues =
Set Text -> [Text]
forall a. Set a -> [a]
Set.toAscList (Set Text -> [Text]) -> ([Text] -> Set Text) -> [Text] -> [Text]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Set Text -> Set Text -> Set Text
forall a. Ord a => Set a -> Set a -> Set a
Set.union ([Text] -> Set Text
forall a. Ord a => [a] -> Set a
Set.fromList [Text]
queues) (Set Text -> Set Text)
-> ([Text] -> Set Text) -> [Text] -> Set Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Text] -> Set Text
forall a. Ord a => [a] -> Set a
Set.fromList
([Text] -> [Text]) -> m [Text] -> m [Text]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> 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.existingQueuesSQL Text
schemaName)
getQueue
:: (MonadArbiter m)
=> SchemaName
-> Text
-> m (Maybe QueueRow)
getQueue :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe QueueRow)
getQueue Text
schemaName Text
queue =
Query QueueRow -> m (Maybe QueueRow)
forall (m :: * -> *) a. MonadArbiter m => Query a -> m (Maybe a)
MA.queryMaybe (Text -> Text -> Query QueueRow
Tmpl.getQueueSQL Text
schemaName Text
queue)
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)