{-# LANGUAGE AllowAmbiguousTypes #-}
{-# LANGUAGE DuplicateRecordFields #-}

-- | Job queue operations over the registry. A payload operation gets its table from the
-- payload's registry entry.
module Arbiter.Core.HighLevel
  ( -- * Constraint aliases
    QueueOperation
  , JobOperation
  , RegistryAdmissionPolicies

    -- * Job operations
  , insertJob
  , insertJobsBatch
  , insertJobsBatch_
  , claimNextVisibleJobs
  , claimNextVisibleJobsAs
  , claimNextVisibleJobsBatched
  , ackJob
  , ackJobsBatch
  , updateJobForRetry
  , nackJob
  , nackJobAfter
  , nackJobsBatch
  , nackJobsBatchAfter
  , setVisibilityTimeout
  , setVisibilityTimeoutBatch
  , SetVisibilityResult (..)

    -- * Rate-limit operations
  , addRateLimitTokens
  , pruneRateLimitBuckets
  , resetRateLimitBuckets
  , listRateLimitPolicies
  , getRateLimitPolicy
  , rateLimitPolicyExists
  , listRateLimitBuckets
  , updateRateLimitPolicyOverrides
  , setRateLimit
  , clearRateLimit

    -- * Concurrency operations
  , listConcurrencyPolicies
  , getConcurrencyPolicy
  , concurrencyPolicyExists
  , listConcurrencyKeys
  , updateConcurrencyPolicyOverrides
  , setConcurrencyLimit
  , clearConcurrencyLimit
  , pruneConcurrencyKeys
  , reconcileConcurrencyCounts
  , reconcileConcurrencyCountsIfStale
  , reconcileAndPruneConcurrency

    -- * Filtered query operations
  , Ops.JobFilter (..)
  , Ops.JobSortColumn (..)
  , Ops.DLQSortColumn (..)
  , Ops.ArchiveSortColumn (..)
  , Ops.SortDir (..)
  , listJobsFiltered
  , countJobsFiltered
  , listDLQFiltered
  , countDLQFiltered
  , listArchiveFiltered
  , countArchiveFiltered

    -- * Dead-letter queue operations
  , moveToDLQ
  , moveToDLQBatch
  , listDLQJobs
  , retryFromDLQ
  , retryFromDLQWithPayload
  , retryFromDLQBatch
  , retryDLQFiltered
  , dlqJobExists
  , deleteDLQJob
  , deleteDLQJobsBatch

    -- * Archive operations
  , listArchiveJobs
  , getArchiveJobById
  , listArchiveJobsByGroup
  , deleteArchiveJob
  , deleteArchiveJobsBatch
  , reEnqueueFromArchive
  , reEnqueueFromArchiveWithPayload

    -- * Admin operations
  , listJobs
  , getJobById
  , jobExists
  , listJobsByGroup
  , listJobsByParent
  , cancelJob
  , cancelJobsBatch
  , forceCancelJob
  , promoteJob
  , rescheduleJob
  , Ops.QueueStats (..)
  , Ops.QueueOverview (..)
  , Ops.GroupSummary (..)
  , getQueueStats

    -- * Count operations
  , countJobs
  , countJobsByGroup
  , countJobsByParent
  , countDLQJobs
  , countChildrenBatch
  , countDLQChildren
  , countDLQChildrenBatch

    -- * Job dependency operations
  , pauseChildren
  , resumeChildren
  , cancelJobCascade

    -- * Suspend and resume operations
  , suspendJob
  , resumeJob

    -- * Child results and parent state
  , insertResult
  , insertResultUnsafe
  , getResultsByParent
  , getDLQChildErrorsByParent
  , readChildResultsRaw
  , Ops.mergeRawChildResults
  , persistParentState
  , getParentStateSnapshot

    -- * Groups table operations
  , refreshAllGroupsFully

    -- * Worker registry operations
  , registerWorker
  , heartbeatWorker
  , setWorkerPaused
  , markWorkerShuttingDown
  , deregisterWorker
  , listWorkers
  , sweepStaleWorkers
  , WorkerRow (..)
  , WorkerHealth (..)

    -- * Queue operations
  , ensureQueue
  , setQueuePaused
  , setQueueDLQRetention
  , getQueue
  , listQueues
  , QueueRow (..)

    -- * Cron schedule operations
  , listCronSchedules
  , getCronScheduleByName
  , updateCronScheduleUnchecked
  , CronScheduleRow (..)
  , CronScheduleUpdate (..)

    -- * Global gate
  , runGated

    -- * Job tree DSL
  , insertJobTree
  , spawnChildren

    -- * Table lookup
  , queueTable

    -- * Re-exports
  , getSchema
  ) where

import Control.Monad (void, when)
import Data.Aeson (Value)
import Data.Foldable (toList)
import Data.Int (Int32, Int64)
import Data.List (genericLength)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Proxy (Proxy (..))
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (NominalDiffTime, UTCTime)
import Data.UUID.Types (UUID)
import GHC.TypeLits (KnownSymbol, symbolVal)
import UnliftIO (MonadUnliftIO)

import Arbiter.Core.Admission (AdmissionPolicy (..))
import Arbiter.Core.Concurrency.Spec (ConcurrencyPolicy (..))
import Arbiter.Core.Concurrency.Stats (ConcurrencyKeyView, ConcurrencyPolicyUpdate (..), ConcurrencyPolicyView)
import Arbiter.Core.CronSchedule (CronScheduleRow (..), CronScheduleUpdate (..))
import Arbiter.Core.Job.Archive qualified as Archive
import Arbiter.Core.Job.DLQ qualified as DLQ
import Arbiter.Core.Job.Schema (SchemaName, TableName)
import Arbiter.Core.Job.Types
  ( ClaimSeq
  , HasKind
  , JobId
  , JobPayload
  , JobRead
  , JobWrite
  , RegistryAdmissionPolicies
  , claimSeq
  , claimedBy
  , kindsFor
  , primaryKey
  )
import Arbiter.Core.Job.Types qualified as Job
import Arbiter.Core.JobResult (EncodeJobResult, encodeJobResult)
import Arbiter.Core.JobTree qualified as JT
import Arbiter.Core.MonadArbiter (MonadArbiter (..), ResultOf)
import Arbiter.Core.Operations qualified as Ops
import Arbiter.Core.QueueRegistry (RegistryTables (..), TableForPayload)
import Arbiter.Core.Queues (QueueRow (..))
import Arbiter.Core.RateLimit.Spec (RateLimitKey (..), RateLimitPolicy (..))
import Arbiter.Core.RateLimit.Stats (RateLimitBucketView, RateLimitPolicyUpdate (..), RateLimitPolicyView)
import Arbiter.Core.Trace (withPublishSpan)
import Arbiter.Core.Worker (WorkerHealth (..), WorkerRow (..))

-- | Constraints for queue operations (requires table name lookup from registry).
type QueueOperation m payload =
  ( JobPayload payload
  , KnownSymbol (TableForPayload payload (RegistryOf m))
  , MonadArbiter m
  )

-- | Constraints for job operations (table name stored in job).
type JobOperation m payload = (JobPayload payload, MonadArbiter m)

-- | The table name @payload@'s entry in this monad's registry declares.
queueTable :: forall payload m. (KnownSymbol (TableForPayload payload (RegistryOf m))) => Text
queueTable :: forall payload (m :: * -> *).
KnownSymbol (TableForPayload payload (RegistryOf m)) =>
Text
queueTable = String -> Text
T.pack (String -> Text) -> String -> Text
forall a b. (a -> b) -> a -> b
$ Proxy (SpecName (MatchIn payload '[] (RegistryOf m))) -> String
forall (n :: Symbol) (proxy :: Symbol -> *).
KnownSymbol n =>
proxy n -> String
symbolVal (forall {k} (t :: k). Proxy t
forall (t :: Symbol). Proxy t
Proxy @(TableForPayload payload (RegistryOf m)))

-- | Run an operation against the schema and the table @payload@'s registry entry names.
onQueue
  :: forall payload m a
   . (KnownSymbol (TableForPayload payload (RegistryOf m)), MonadArbiter m)
  => (SchemaName -> TableName -> m a)
  -> m a
onQueue :: forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue Text -> Text -> m a
operation = m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema m Text -> (Text -> m a) -> m a
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Text
schemaName -> Text -> Text -> m a
operation Text
schemaName (forall payload (m :: * -> *).
KnownSymbol (TableForPayload payload (RegistryOf m)) =>
Text
queueTable @payload @m)

-- | Run an operation against the schema.
onSchema :: (MonadArbiter m) => (SchemaName -> m a) -> m a
onSchema :: forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema = (m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema m Text -> (Text -> m a) -> m a
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>=)

-- | Run an operation against the schema and every table the registry names.
onRegistry
  :: forall m a
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => (SchemaName -> [TableName] -> m a)
  -> m a
onRegistry :: forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry Text -> [Text] -> m a
operation = m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema m Text -> (Text -> m a) -> m a
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Text
schemaName -> Text -> [Text] -> m a
operation Text
schemaName (Proxy (RegistryOf m) -> [Text]
forall (registry :: JobPayloadRegistry).
RegistryTables registry =>
Proxy registry -> [Text]
registryTableNames (forall (t :: JobPayloadRegistry). Proxy t
forall {k} (t :: k). Proxy t
Proxy @(RegistryOf m)))

-- | Run an operation against the schema and the table a job came from.
onJob :: (MonadArbiter m) => JobRead payload -> (SchemaName -> TableName -> m a) -> m a
onJob :: forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job Text -> Text -> m a
operation = m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema m Text -> (Text -> m a) -> m a
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Text
schemaName -> Text -> Text -> m a
operation Text
schemaName (JobRead payload -> Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> q
Job.queueName JobRead payload
job)

-- | 'onJob' keyed by a batch's first job, or @none@ for an empty batch.
onJobs :: (MonadArbiter m) => (item -> JobRead payload) -> a -> [item] -> (SchemaName -> TableName -> m a) -> m a
onJobs :: forall (m :: * -> *) item payload a.
MonadArbiter m =>
(item -> JobRead payload)
-> a -> [item] -> (Text -> Text -> m a) -> m a
onJobs item -> JobRead payload
_ a
none [] Text -> Text -> m a
_ = a -> m a
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure a
none
onJobs item -> JobRead payload
jobOf a
_ (item
item : [item]
_) Text -> Text -> m a
operation = JobRead payload -> (Text -> Text -> m a) -> m a
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob (item -> JobRead payload
jobOf item
item) Text -> Text -> m a
operation

publishSpan
  :: forall payload m a
   . (HasKind payload, KnownSymbol (TableForPayload payload (RegistryOf m)), MonadUnliftIO m)
  => [JobWrite payload]
  -> m a
  -> m a
publishSpan :: forall payload (m :: * -> *) a.
(HasKind payload,
 KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadUnliftIO m) =>
[JobWrite payload] -> m a -> m a
publishSpan = Text -> [JobWrite payload] -> m a -> m a
forall payload (m :: * -> *) a.
(HasKind payload, MonadUnliftIO m) =>
Text -> [JobWrite payload] -> m a -> m a
withPublishSpan (forall payload (m :: * -> *).
KnownSymbol (TableForPayload payload (RegistryOf m)) =>
Text
queueTable @payload @m)

-- | Insert a job. Returns the job inserted or replaced. 'Nothing' when an
-- 'Arbiter.Core.Job.Dedup.IgnoreDuplicate' key already exists, or a
-- 'Arbiter.Core.Job.Dedup.ReplaceDuplicate' key names a job that is in flight (claimed
-- under a live lease), force-cancel-flagged, has a different parent, or has children
-- (DLQ children count).
insertJob
  :: forall payload m
   . (QueueOperation m payload)
  => JobWrite payload
  -> m (Maybe (JobRead payload))
insertJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
JobWrite payload -> m (Maybe (JobRead payload))
insertJob JobWrite payload
job =
  forall payload (m :: * -> *) a.
(HasKind payload,
 KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadUnliftIO m) =>
[JobWrite payload] -> m a -> m a
publishSpan @payload [JobWrite payload
job] (m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe (JobRead payload)))
 -> m (Maybe (JobRead payload)))
-> (Text -> Text -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> JobWrite payload -> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> JobWrite payload -> m (Maybe (JobRead payload))
Ops.insertJob Text
schemaName Text
tableName JobWrite payload
job

-- | Insert multiple jobs in one round-trip. Returns the jobs inserted or replaced. An
-- ignored duplicate is left out. Use 'insertJobTree' for parent-child relationships.
insertJobsBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [JobWrite payload]
  -> m [JobRead payload]
insertJobsBatch :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m [JobRead payload]
insertJobsBatch [] = [JobRead payload] -> m [JobRead payload]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
insertJobsBatch [JobWrite payload]
jobs =
  forall payload (m :: * -> *) a.
(HasKind payload,
 KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadUnliftIO m) =>
[JobWrite payload] -> m a -> m a
publishSpan @payload [JobWrite payload]
jobs (m [JobRead payload] -> m [JobRead payload])
-> m [JobRead payload] -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [JobRead payload]) -> m [JobRead payload])
-> (Text -> Text -> m [JobRead payload]) -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> [JobWrite payload] -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobWrite payload] -> m [JobRead payload]
Ops.insertJobsBatch Text
schemaName Text
tableName [JobWrite payload]
jobs

-- | 'insertJobsBatch' returning the rows affected, dedup replacements included.
insertJobsBatch_
  :: forall payload m
   . (QueueOperation m payload)
  => [JobWrite payload]
  -> m Int64
insertJobsBatch_ :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobWrite payload] -> m Int64
insertJobsBatch_ [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
insertJobsBatch_ [JobWrite payload]
jobs =
  forall payload (m :: * -> *) a.
(HasKind payload,
 KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadUnliftIO m) =>
[JobWrite payload] -> m a -> m a
publishSpan @payload [JobWrite payload]
jobs (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> [JobWrite payload] -> m Int64
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobWrite payload] -> m Int64
Ops.insertJobsBatch_ Text
schemaName Text
tableName [JobWrite payload]
jobs

-- | Claim visible jobs, at most one per group, and fewer than the limit once the groups
-- run out. Stamps the anonymous claimant. A concurrency policy caps this path as it
-- caps any other claim. A row whose payload fails to decode moves to the DLQ.
claimNextVisibleJobs
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -- ^ Maximum number of jobs to claim.
  -> NominalDiffTime
  -- ^ How long the claimed jobs should remain invisible (in seconds).
  -> m [JobRead payload]
claimNextVisibleJobs :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
claimNextVisibleJobs Int
limit NominalDiffTime
timeout =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [JobRead payload]) -> m [JobRead payload])
-> (Text -> Text -> m [JobRead payload]) -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int -> NominalDiffTime -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> NominalDiffTime -> m [JobRead payload]
Ops.claimNextVisibleJobs Text
schemaName Text
tableName Int
limit NominalDiffTime
timeout

-- | 'claimNextVisibleJobs' under a given worker id. A row whose payload fails to decode
-- moves to the DLQ.
claimNextVisibleJobsAs
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -- ^ Maximum number of jobs to claim.
  -> NominalDiffTime
  -- ^ How long the claimed jobs should remain invisible (in seconds).
  -> UUID
  -- ^ Worker UUID stamped on each claimed row.
  -> m [JobRead payload]
claimNextVisibleJobsAs :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> UUID -> m [JobRead payload]
claimNextVisibleJobsAs Int
limit NominalDiffTime
timeout UUID
workerId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [JobRead payload]) -> m [JobRead payload])
-> (Text -> Text -> m [JobRead payload]) -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text -> Int -> NominalDiffTime -> UUID -> m [JobRead payload]
Ops.claimNextVisibleJobsAs Text
schemaName Text
tableName Int
limit NominalDiffTime
timeout UUID
workerId

-- | 'claimNextVisibleJobs' claiming up to @batchSize@ jobs from each group, still in
-- per-group order. Stamps the anonymous claimant. A row whose payload fails to decode
-- moves to the DLQ.
claimNextVisibleJobsBatched
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -- ^ Batch size: maximum number of jobs to claim per group.
  -> Int
  -- ^ Maximum number of batches to claim. Each batch is one group, or ungrouped jobs
  -- batched together.
  -> NominalDiffTime
  -- ^ How long the claimed jobs should remain invisible (in seconds).
  -> m [NonEmpty (JobRead payload)]
claimNextVisibleJobsBatched :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> NominalDiffTime -> m [NonEmpty (JobRead payload)]
claimNextVisibleJobsBatched Int
batchSize Int
maxBatches NominalDiffTime
timeout =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [NonEmpty (JobRead payload)])
 -> m [NonEmpty (JobRead payload)])
-> (Text -> Text -> m [NonEmpty (JobRead payload)])
-> m [NonEmpty (JobRead payload)]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> m [NonEmpty (JobRead payload)]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> m [NonEmpty (JobRead payload)]
Ops.claimNextVisibleJobsBatched Text
schemaName Text
tableName Int
batchSize Int
maxBatches NominalDiffTime
timeout

-- | Ack a completed job, deleting it or suspending it when its children are still
-- running. Returns 1, or 0 for a job that is gone or reclaimed.
ackJob
  :: forall payload m
   . (MonadArbiter m)
  => JobRead payload
  -> m Int64
ackJob :: forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
ackJob JobRead payload
job = JobRead payload -> (Text -> Text -> m Int64) -> m Int64
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> JobRead payload -> m Int64
Ops.ackJob Text
schemaName Text
tableName JobRead payload
job

-- | 'ackJob' over a batch from one queue in one statement, returning the ids acked.
-- Gone or reclaimed jobs are absent.
ackJobsBatch
  :: forall payload m
   . (MonadArbiter m)
  => [JobRead payload]
  -> m [Int64]
ackJobsBatch :: forall payload (m :: * -> *).
MonadArbiter m =>
[JobRead payload] -> m [Int64]
ackJobsBatch [JobRead payload]
jobs = (JobRead payload -> JobRead payload)
-> [Int64]
-> [JobRead payload]
-> (Text -> Text -> m [Int64])
-> m [Int64]
forall (m :: * -> *) item payload a.
MonadArbiter m =>
(item -> JobRead payload)
-> a -> [item] -> (Text -> Text -> m a) -> m a
onJobs JobRead payload -> JobRead payload
forall a. a -> a
id [] [JobRead payload]
jobs ((Text -> Text -> m [Int64]) -> m [Int64])
-> (Text -> Text -> m [Int64]) -> m [Int64]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [JobRead payload] -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [JobRead payload] -> m [Int64]
Ops.ackJobsBatch Text
schemaName Text
tableName [JobRead payload]
jobs

-- | Park a failed job for its retry backoff. Returns 0 for a job another worker holds or
-- a suspended job.
updateJobForRetry
  :: forall payload m
   . (MonadArbiter m)
  => NominalDiffTime
  -- ^ The delay before the retry, rounded up to whole seconds.
  -> Text
  -- ^ An error message to store with the job.
  -> JobRead payload
  -> m Int64
updateJobForRetry :: forall payload (m :: * -> *).
MonadArbiter m =>
NominalDiffTime -> Text -> JobRead payload -> m Int64
updateJobForRetry NominalDiffTime
delay Text
errorMsg JobRead payload
job =
  JobRead payload -> (Text -> Text -> m Int64) -> m Int64
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text
-> Text -> NominalDiffTime -> Text -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> NominalDiffTime -> Text -> JobRead payload -> m Int64
Ops.updateJobForRetry Text
schemaName Text
tableName NominalDiffTime
delay Text
errorMsg JobRead payload
job

-- | Soft-nack a job. It is reprocessed once its visibility timeout lapses. No failure
-- is recorded and no attempt is consumed. Returns 0 for a job another worker holds, a
-- suspended job, or a released claim.
nackJob
  :: forall payload m
   . (MonadArbiter m)
  => JobRead payload
  -> m Int64
nackJob :: forall payload (m :: * -> *).
MonadArbiter m =>
JobRead payload -> m Int64
nackJob JobRead payload
job = JobRead payload -> (Text -> Text -> m Int64) -> m Int64
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Maybe NominalDiffTime -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Maybe NominalDiffTime -> JobRead payload -> m Int64
Ops.nackJob Text
schemaName Text
tableName Maybe NominalDiffTime
forall a. Maybe a
Nothing JobRead payload
job

-- | 'nackJob' that reprocesses the job after a delay instead of at the lease's end.
nackJobAfter
  :: forall payload m
   . (MonadArbiter m)
  => NominalDiffTime
  -- ^ The delay before the job is visible again, rounded up to whole seconds.
  -> JobRead payload
  -> m Int64
nackJobAfter :: forall payload (m :: * -> *).
MonadArbiter m =>
NominalDiffTime -> JobRead payload -> m Int64
nackJobAfter NominalDiffTime
delay JobRead payload
job = JobRead payload -> (Text -> Text -> m Int64) -> m Int64
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Maybe NominalDiffTime -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Maybe NominalDiffTime -> JobRead payload -> m Int64
Ops.nackJob Text
schemaName Text
tableName (NominalDiffTime -> Maybe NominalDiffTime
forall a. a -> Maybe a
Just NominalDiffTime
delay) JobRead payload
job

-- | 'nackJob' over a batch from one queue in one statement, returning the ids nacked.
-- Jobs another worker holds, suspended jobs and released claims are absent.
nackJobsBatch
  :: forall payload m
   . (MonadArbiter m)
  => [JobRead payload]
  -> m [Int64]
nackJobsBatch :: forall payload (m :: * -> *).
MonadArbiter m =>
[JobRead payload] -> m [Int64]
nackJobsBatch [JobRead payload]
jobs = (JobRead payload -> JobRead payload)
-> [Int64]
-> [JobRead payload]
-> (Text -> Text -> m [Int64])
-> m [Int64]
forall (m :: * -> *) item payload a.
MonadArbiter m =>
(item -> JobRead payload)
-> a -> [item] -> (Text -> Text -> m a) -> m a
onJobs JobRead payload -> JobRead payload
forall a. a -> a
id [] [JobRead payload]
jobs ((Text -> Text -> m [Int64]) -> m [Int64])
-> (Text -> Text -> m [Int64]) -> m [Int64]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text
-> Text -> Maybe NominalDiffTime -> [JobRead payload] -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Maybe NominalDiffTime -> [JobRead payload] -> m [Int64]
Ops.nackJobsBatch Text
schemaName Text
tableName Maybe NominalDiffTime
forall a. Maybe a
Nothing [JobRead payload]
jobs

-- | 'nackJobsBatch' that reprocesses the jobs after a delay instead of at the lease's end.
nackJobsBatchAfter
  :: forall payload m
   . (MonadArbiter m)
  => NominalDiffTime
  -- ^ The delay before the jobs are visible again, rounded up to whole seconds.
  -> [JobRead payload]
  -> m [Int64]
nackJobsBatchAfter :: forall payload (m :: * -> *).
MonadArbiter m =>
NominalDiffTime -> [JobRead payload] -> m [Int64]
nackJobsBatchAfter NominalDiffTime
delay [JobRead payload]
jobs =
  (JobRead payload -> JobRead payload)
-> [Int64]
-> [JobRead payload]
-> (Text -> Text -> m [Int64])
-> m [Int64]
forall (m :: * -> *) item payload a.
MonadArbiter m =>
(item -> JobRead payload)
-> a -> [item] -> (Text -> Text -> m a) -> m a
onJobs JobRead payload -> JobRead payload
forall a. a -> a
id [] [JobRead payload]
jobs ((Text -> Text -> m [Int64]) -> m [Int64])
-> (Text -> Text -> m [Int64]) -> m [Int64]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text
-> Text -> Maybe NominalDiffTime -> [JobRead payload] -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Maybe NominalDiffTime -> [JobRead payload] -> m [Int64]
Ops.nackJobsBatch Text
schemaName Text
tableName (NominalDiffTime -> Maybe NominalDiffTime
forall a. a -> Maybe a
Just NominalDiffTime
delay) [JobRead payload]
jobs

-- | Extend a job's visibility timeout by hand, for a long-running job. Matches the claim
-- token only. It also extends a job the caller released (nack, retry). Returns 0 for a
-- job that is gone, reclaimed or suspended.
setVisibilityTimeout
  :: forall payload m
   . (MonadArbiter m)
  => NominalDiffTime
  -- ^ The new visibility timeout (in seconds) from the current time.
  -> JobRead payload
  -> m Int64
setVisibilityTimeout :: forall payload (m :: * -> *).
MonadArbiter m =>
NominalDiffTime -> JobRead payload -> m Int64
setVisibilityTimeout NominalDiffTime
timeout JobRead payload
job =
  JobRead payload -> (Text -> Text -> m Int64) -> m Int64
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> NominalDiffTime -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> NominalDiffTime -> JobRead payload -> m Int64
Ops.setVisibilityTimeout Text
schemaName Text
tableName NominalDiffTime
timeout JobRead payload
job

-- | Extend the visibility timeout of a batch from one queue. Each row must still match
-- its holder. A row whose row lock or group lock is busy is skipped.
setVisibilityTimeoutBatch
  :: forall payload m
   . (MonadArbiter m)
  => NominalDiffTime
  -- ^ The new visibility timeout (in seconds) from the current time.
  -> [JobRead payload]
  -- ^ Jobs to heartbeat (all must be from the same queue)
  -> m [SetVisibilityResult]
setVisibilityTimeoutBatch :: forall payload (m :: * -> *).
MonadArbiter m =>
NominalDiffTime -> [JobRead payload] -> m [SetVisibilityResult]
setVisibilityTimeoutBatch NominalDiffTime
timeout [JobRead payload]
jobs =
  (JobRead payload -> JobRead payload)
-> [SetVisibilityResult]
-> [JobRead payload]
-> (Text -> Text -> m [SetVisibilityResult])
-> m [SetVisibilityResult]
forall (m :: * -> *) item payload a.
MonadArbiter m =>
(item -> JobRead payload)
-> a -> [item] -> (Text -> Text -> m a) -> m a
onJobs JobRead payload -> JobRead payload
forall a. a -> a
id [] [JobRead payload]
jobs ((Text -> Text -> m [SetVisibilityResult])
 -> m [SetVisibilityResult])
-> (Text -> Text -> m [SetVisibilityResult])
-> m [SetVisibilityResult]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    (VisibilityUpdateInfo -> SetVisibilityResult)
-> [VisibilityUpdateInfo] -> [SetVisibilityResult]
forall a b. (a -> b) -> [a] -> [b]
map VisibilityUpdateInfo -> SetVisibilityResult
toResult ([VisibilityUpdateInfo] -> [SetVisibilityResult])
-> m [VisibilityUpdateInfo] -> m [SetVisibilityResult]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text
-> Text
-> NominalDiffTime
-> [JobRead payload]
-> m [VisibilityUpdateInfo]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> NominalDiffTime
-> [JobRead payload]
-> m [VisibilityUpdateInfo]
Ops.setVisibilityTimeoutBatch Text
schemaName Text
tableName NominalDiffTime
timeout [JobRead payload]
jobs
  where
    jobMap :: Map Int64 (JobRead payload)
jobMap = [(Int64, JobRead payload)] -> Map Int64 (JobRead payload)
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList [(JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
primaryKey JobRead payload
job, JobRead payload
job) | JobRead payload
job <- [JobRead payload]
jobs]
    toResult :: VisibilityUpdateInfo -> SetVisibilityResult
toResult
      Ops.VisibilityUpdateInfo
        { vuiJobId :: VisibilityUpdateInfo -> Int64
Ops.vuiJobId = Int64
jobId
        , vuiWasHeartbeated :: VisibilityUpdateInfo -> Bool
Ops.vuiWasHeartbeated = Bool
heartbeated
        , vuiCurrentDbClaimSeq :: VisibilityUpdateInfo -> Maybe Int64
Ops.vuiCurrentDbClaimSeq = Maybe Int64
mActual
        , vuiCancelRequested :: VisibilityUpdateInfo -> Bool
Ops.vuiCancelRequested = Bool
cancelled
        , vuiSuspended :: VisibilityUpdateInfo -> Bool
Ops.vuiSuspended = Bool
suspended
        , vuiClaimedBy :: VisibilityUpdateInfo -> Maybe UUID
Ops.vuiClaimedBy = Maybe UUID
holder
        } =
        let mJob :: Maybe (JobRead payload)
mJob = Int64 -> Map Int64 (JobRead payload) -> Maybe (JobRead payload)
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Int64
jobId Map Int64 (JobRead payload)
jobMap
            expected :: Int64
expected = Int64
-> (JobRead payload -> Int64) -> Maybe (JobRead payload) -> Int64
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Int64
0 JobRead payload -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
claimSeq Maybe (JobRead payload)
mJob
            heldHere :: Bool
heldHere = (JobRead payload -> Bool) -> Maybe (JobRead payload) -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any ((Maybe UUID -> Maybe UUID -> Bool
forall a. Eq a => a -> a -> Bool
== Maybe UUID
holder) (Maybe UUID -> Bool)
-> (JobRead payload -> Maybe UUID) -> JobRead payload -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. JobRead payload -> Maybe UUID
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe UUID
claimedBy) Maybe (JobRead payload)
mJob
         in case Maybe Int64
mActual of
              Maybe Int64
Nothing -> Int64 -> SetVisibilityResult
JobGone Int64
jobId
              Just Int64
actual
                | Bool
cancelled, Bool
heldHere, Int64
actual Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
== Int64
expected Int64 -> Int64 -> Int64
forall a. Num a => a -> a -> a
+ Int64
Ops.cancelTokenBump -> Int64 -> SetVisibilityResult
JobCancelled Int64
jobId
                | Int64
actual Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
/= Int64
expected -> Int64 -> Int64 -> Int64 -> SetVisibilityResult
JobReclaimed Int64
jobId Int64
expected Int64
actual
                | Bool
suspended -> Int64 -> SetVisibilityResult
JobSuspended Int64
jobId
                | Bool
heartbeated -> Int64 -> SetVisibilityResult
VisibilityExtended Int64
jobId
                | Bool
otherwise -> Int64 -> SetVisibilityResult
VisibilityUnchanged Int64
jobId

-- | Result of setting visibility timeout for a single job in a batch.
data SetVisibilityResult
  = -- | Visibility timeout was successfully extended.
    VisibilityExtended JobId
  | -- | Job no longer exists (was deleted/acked).
    JobGone JobId
  | -- | Job was reclaimed by another worker. Carries this claim's token, then the
    -- row's current token.
    JobReclaimed JobId ClaimSeq ClaimSeq
  | -- | Job was force-cancel-flagged.
    JobCancelled JobId
  | -- | Job is suspended. It holds no lease.
    JobSuspended JobId
  | -- | Nothing was extended. The claim was released, the row or its group summary was
    -- locked, the row changed mid-statement, or another claimant holds it at the same
    -- claim token.
    VisibilityUnchanged JobId
  deriving stock (SetVisibilityResult -> SetVisibilityResult -> Bool
(SetVisibilityResult -> SetVisibilityResult -> Bool)
-> (SetVisibilityResult -> SetVisibilityResult -> Bool)
-> Eq SetVisibilityResult
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: SetVisibilityResult -> SetVisibilityResult -> Bool
== :: SetVisibilityResult -> SetVisibilityResult -> Bool
$c/= :: SetVisibilityResult -> SetVisibilityResult -> Bool
/= :: SetVisibilityResult -> SetVisibilityResult -> Bool
Eq, Int -> SetVisibilityResult -> ShowS
[SetVisibilityResult] -> ShowS
SetVisibilityResult -> String
(Int -> SetVisibilityResult -> ShowS)
-> (SetVisibilityResult -> String)
-> ([SetVisibilityResult] -> ShowS)
-> Show SetVisibilityResult
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SetVisibilityResult -> ShowS
showsPrec :: Int -> SetVisibilityResult -> ShowS
$cshow :: SetVisibilityResult -> String
show :: SetVisibilityResult -> String
$cshowList :: [SetVisibilityResult] -> ShowS
showList :: [SetVisibilityResult] -> ShowS
Show)

-- | Add tokens to a key's bucket, clamped to @[0, max]@, and wake any of its throttled
-- jobs. A key without a bucket gets a full bucket and the amount has no effect. Without a
-- policy the bucket stays unchanged and the wake still runs. Returns the number of jobs
-- woken.
addRateLimitTokens
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => RateLimitKey
  -> Double
  -> m Int64
addRateLimitTokens :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
RateLimitKey -> Double -> m Int64
addRateLimitTokens RateLimitKey
key Double
amount =
  m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ (Text -> [Text] -> m Int64) -> m Int64
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry ((Text -> [Text] -> m Int64) -> m Int64)
-> (Text -> [Text] -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName [Text]
queues ->
    Text -> RateLimitKey -> Double -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> RateLimitKey -> Double -> m ()
Ops.addRateLimitTokens Text
schemaName RateLimitKey
key Double
amount m () -> m Int64 -> m Int64
forall a b. m a -> m b -> m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Text -> [Text] -> RateLimitKey -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> RateLimitKey -> m Int64
Ops.wakeThrottledJobsForKey Text
schemaName [Text]
queues RateLimitKey
key

-- | Delete reclaimable idle (full) buckets. Returns the number pruned. The worker
-- pool reaper runs this. A full bucket re-seeds at full on next use.
pruneRateLimitBuckets
  :: forall m
   . (MonadArbiter m)
  => NominalDiffTime
  -> m Int64
pruneRateLimitBuckets :: forall (m :: * -> *). MonadArbiter m => NominalDiffTime -> m Int64
pruneRateLimitBuckets NominalDiffTime
idle = (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> NominalDiffTime -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> NominalDiffTime -> m Int64
Ops.pruneRateLimitBuckets Text
schemaName NominalDiffTime
idle

-- | Reset every bucket for a prefix to full and wake its throttled jobs. Returns
-- the number of buckets reset. A manual (0-refill) policy plus a cron calling this
-- at the boundary is a fixed window.
resetRateLimitBuckets
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => Text
  -> m Int64
resetRateLimitBuckets :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> m Int64
resetRateLimitBuckets Text
prefix =
  m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ (Text -> [Text] -> m Int64) -> m Int64
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry ((Text -> [Text] -> m Int64) -> m Int64)
-> (Text -> [Text] -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName [Text]
queues ->
    Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
Ops.resetRateLimitBuckets Text
schemaName Text
prefix m Int64 -> m Int64 -> m Int64
forall a b. m a -> m b -> m a
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f a
<* Text -> [Text] -> Text -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> Text -> m Int64
Ops.wakeThrottledJobs Text
schemaName [Text]
queues Text
prefix

-- | List every rate-limit policy with its params, bucket stats, and a live count
-- of currently-throttled jobs per prefix across the registry's queues.
listRateLimitPolicies
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m [RateLimitPolicyView]
listRateLimitPolicies :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m [RateLimitPolicyView]
listRateLimitPolicies = (Text -> [Text] -> m [RateLimitPolicyView])
-> m [RateLimitPolicyView]
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry Text -> [Text] -> m [RateLimitPolicyView]
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m [RateLimitPolicyView]
Ops.listRateLimitPolicies

-- | One prefix's rate-limit policy with its params, bucket stats, and live throttled
-- count. 'Nothing' when the prefix has no policy.
getRateLimitPolicy
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => Text
  -> m (Maybe RateLimitPolicyView)
getRateLimitPolicy :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> m (Maybe RateLimitPolicyView)
getRateLimitPolicy Text
prefix = (Text -> [Text] -> m (Maybe RateLimitPolicyView))
-> m (Maybe RateLimitPolicyView)
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry ((Text -> [Text] -> m (Maybe RateLimitPolicyView))
 -> m (Maybe RateLimitPolicyView))
-> (Text -> [Text] -> m (Maybe RateLimitPolicyView))
-> m (Maybe RateLimitPolicyView)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName [Text]
queues -> Text -> [Text] -> Text -> m (Maybe RateLimitPolicyView)
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> Text -> m (Maybe RateLimitPolicyView)
Ops.getRateLimitPolicy Text
schemaName [Text]
queues Text
prefix

-- | Whether a rate-limit policy exists for a prefix.
rateLimitPolicyExists
  :: forall m
   . (MonadArbiter m)
  => Text
  -> m Bool
rateLimitPolicyExists :: forall (m :: * -> *). MonadArbiter m => Text -> m Bool
rateLimitPolicyExists Text
prefix = (Text -> m Bool) -> m Bool
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Bool) -> m Bool) -> (Text -> m Bool) -> m Bool
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> m Bool
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Bool
Ops.rateLimitPolicyExists Text
schemaName Text
prefix

-- | List a prefix's buckets with fill levels, emptiest first.
listRateLimitBuckets
  :: forall m
   . (MonadArbiter m)
  => Text
  -- ^ Prefix
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [RateLimitBucketView]
listRateLimitBuckets :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Int -> Int -> m [RateLimitBucketView]
listRateLimitBuckets Text
prefix Int
limit Int
offset =
  (Text -> m [RateLimitBucketView]) -> m [RateLimitBucketView]
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m [RateLimitBucketView]) -> m [RateLimitBucketView])
-> (Text -> m [RateLimitBucketView]) -> m [RateLimitBucketView]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
    Text -> Text -> Int -> Int -> m [RateLimitBucketView]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [RateLimitBucketView]
Ops.listRateLimitBuckets Text
schemaName Text
prefix Int
limit Int
offset

-- | Set or clear a policy's override params and wake the prefix's throttled jobs. Returns
-- rows affected (0 if absent).
updateRateLimitPolicyOverrides
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => Text
  -> RateLimitPolicyUpdate
  -> m Int64
updateRateLimitPolicyOverrides :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides Text
prefix RateLimitPolicyUpdate
upd = (Text -> [Text] -> m Int64) -> m Int64
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry ((Text -> [Text] -> m Int64) -> m Int64)
-> (Text -> [Text] -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName [Text]
queues -> do
  affected <- Text -> Text -> RateLimitPolicyUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> RateLimitPolicyUpdate -> m Int64
Ops.updateRateLimitPolicyOverrides Text
schemaName Text
prefix RateLimitPolicyUpdate
upd
  -- Wake after the override commits.
  when (affected > 0) $ void $ Ops.wakeThrottledJobs schemaName queues prefix
  pure affected

-- | Override a policy with this shape, until it is cleared. Returns rows affected.
setRateLimit :: (MonadArbiter m, RegistryTables (RegistryOf m)) => RateLimitPolicy -> m Int64
setRateLimit :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
RateLimitPolicy -> m Int64
setRateLimit RateLimitPolicy
policy =
  Text -> RateLimitPolicyUpdate -> m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides (RateLimitPolicy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf RateLimitPolicy
policy) (RateLimitPolicyUpdate -> m Int64)
-> RateLimitPolicyUpdate -> m Int64
forall a b. (a -> b) -> a -> b
$
    RateLimitPolicyUpdate
      { overrideMaxTokens :: Maybe (Maybe Double)
overrideMaxTokens = Maybe Double -> Maybe (Maybe Double)
forall a. a -> Maybe a
Just (Double -> Maybe Double
forall a. a -> Maybe a
Just (RateLimitPolicy -> Double
policyMax RateLimitPolicy
policy))
      , overrideRefillAmount :: Maybe (Maybe Double)
overrideRefillAmount = Maybe Double -> Maybe (Maybe Double)
forall a. a -> Maybe a
Just (Double -> Maybe Double
forall a. a -> Maybe a
Just (RateLimitPolicy -> Double
policyRefill RateLimitPolicy
policy))
      , overrideInterval :: Maybe (Maybe Double)
overrideInterval = Maybe Double -> Maybe (Maybe Double)
forall a. a -> Maybe a
Just (Double -> Maybe Double
forall a. a -> Maybe a
Just (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (RateLimitPolicy -> NominalDiffTime
policyInterval RateLimitPolicy
policy)))
      }

-- | Drop a policy's overrides. Its declared params apply again. Returns rows affected.
clearRateLimit :: (MonadArbiter m, RegistryTables (RegistryOf m)) => RateLimitPolicy -> m Int64
clearRateLimit :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
RateLimitPolicy -> m Int64
clearRateLimit RateLimitPolicy
policy =
  Text -> RateLimitPolicyUpdate -> m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides (RateLimitPolicy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf RateLimitPolicy
policy) (RateLimitPolicyUpdate -> m Int64)
-> RateLimitPolicyUpdate -> m Int64
forall a b. (a -> b) -> a -> b
$
    RateLimitPolicyUpdate
      { overrideMaxTokens :: Maybe (Maybe Double)
overrideMaxTokens = Maybe Double -> Maybe (Maybe Double)
forall a. a -> Maybe a
Just Maybe Double
forall a. Maybe a
Nothing
      , overrideRefillAmount :: Maybe (Maybe Double)
overrideRefillAmount = Maybe Double -> Maybe (Maybe Double)
forall a. a -> Maybe a
Just Maybe Double
forall a. Maybe a
Nothing
      , overrideInterval :: Maybe (Maybe Double)
overrideInterval = Maybe Double -> Maybe (Maybe Double)
forall a. a -> Maybe a
Just Maybe Double
forall a. Maybe a
Nothing
      }

-- | List every concurrency policy with its default\/override limit and live key and
-- in-flight aggregates.
listConcurrencyPolicies
  :: forall m
   . (MonadArbiter m)
  => m [ConcurrencyPolicyView]
listConcurrencyPolicies :: forall (m :: * -> *). MonadArbiter m => m [ConcurrencyPolicyView]
listConcurrencyPolicies = (Text -> m [ConcurrencyPolicyView]) -> m [ConcurrencyPolicyView]
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema Text -> m [ConcurrencyPolicyView]
forall (m :: * -> *).
MonadArbiter m =>
Text -> m [ConcurrencyPolicyView]
Ops.listConcurrencyPolicies

-- | One prefix's concurrency policy with its default\/override limit and live aggregates.
-- 'Nothing' when the prefix has no policy.
getConcurrencyPolicy
  :: forall m
   . (MonadArbiter m)
  => Text
  -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy :: forall (m :: * -> *).
MonadArbiter m =>
Text -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy Text
prefix = (Text -> m (Maybe ConcurrencyPolicyView))
-> m (Maybe ConcurrencyPolicyView)
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m (Maybe ConcurrencyPolicyView))
 -> m (Maybe ConcurrencyPolicyView))
-> (Text -> m (Maybe ConcurrencyPolicyView))
-> m (Maybe ConcurrencyPolicyView)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> m (Maybe ConcurrencyPolicyView)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe ConcurrencyPolicyView)
Ops.getConcurrencyPolicy Text
schemaName Text
prefix

-- | Whether a concurrency policy exists for a prefix.
concurrencyPolicyExists
  :: forall m
   . (MonadArbiter m)
  => Text
  -> m Bool
concurrencyPolicyExists :: forall (m :: * -> *). MonadArbiter m => Text -> m Bool
concurrencyPolicyExists Text
prefix = (Text -> m Bool) -> m Bool
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Bool) -> m Bool) -> (Text -> m Bool) -> m Bool
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> m Bool
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Bool
Ops.concurrencyPolicyExists Text
schemaName Text
prefix

-- | List a prefix's keys with effective cap and in-flight fill fraction, fullest first.
listConcurrencyKeys
  :: forall m
   . (MonadArbiter m)
  => Text
  -- ^ Prefix
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [ConcurrencyKeyView]
listConcurrencyKeys :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Int -> Int -> m [ConcurrencyKeyView]
listConcurrencyKeys Text
prefix Int
limit Int
offset =
  (Text -> m [ConcurrencyKeyView]) -> m [ConcurrencyKeyView]
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m [ConcurrencyKeyView]) -> m [ConcurrencyKeyView])
-> (Text -> m [ConcurrencyKeyView]) -> m [ConcurrencyKeyView]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
    Text -> Text -> Int -> Int -> m [ConcurrencyKeyView]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [ConcurrencyKeyView]
Ops.listConcurrencyKeys Text
schemaName Text
prefix Int
limit Int
offset

-- | Apply a concurrency policy's override-limit patch on its policy row, retuning every key under
-- the prefix live. Lowering it never preempts in-flight jobs.
-- Returns rows affected.
updateConcurrencyPolicyOverrides
  :: forall m
   . (MonadArbiter m)
  => Text
  -> ConcurrencyPolicyUpdate
  -> m Int64
updateConcurrencyPolicyOverrides :: forall (m :: * -> *).
MonadArbiter m =>
Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides Text
prefix ConcurrencyPolicyUpdate
upd =
  (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
    Text -> Text -> ConcurrencyPolicyUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> ConcurrencyPolicyUpdate -> m Int64
Ops.updateConcurrencyPolicyOverrides Text
schemaName Text
prefix ConcurrencyPolicyUpdate
upd

-- | Override the policy's limit with its 'cpLimit', until it is cleared. Returns rows affected.
setConcurrencyLimit :: (MonadArbiter m) => ConcurrencyPolicy -> m Int64
setConcurrencyLimit :: forall (m :: * -> *).
MonadArbiter m =>
ConcurrencyPolicy -> m Int64
setConcurrencyLimit ConcurrencyPolicy
policy =
  Text -> ConcurrencyPolicyUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides
    (ConcurrencyPolicy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf ConcurrencyPolicy
policy)
    ConcurrencyPolicyUpdate {overrideLimit :: Maybe (Maybe Int32)
overrideLimit = Maybe Int32 -> Maybe (Maybe Int32)
forall a. a -> Maybe a
Just (Int32 -> Maybe Int32
forall a. a -> Maybe a
Just (ConcurrencyPolicy -> Int32
cpLimit ConcurrencyPolicy
policy))}

-- | Drop a concurrency policy's override. Its declared limit applies again. Returns rows affected.
clearConcurrencyLimit :: (MonadArbiter m) => ConcurrencyPolicy -> m Int64
clearConcurrencyLimit :: forall (m :: * -> *).
MonadArbiter m =>
ConcurrencyPolicy -> m Int64
clearConcurrencyLimit ConcurrencyPolicy
policy =
  Text -> ConcurrencyPolicyUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides (ConcurrencyPolicy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf ConcurrencyPolicy
policy) ConcurrencyPolicyUpdate {overrideLimit :: Maybe (Maybe Int32)
overrideLimit = Maybe Int32 -> Maybe (Maybe Int32)
forall a. a -> Maybe a
Just Maybe Int32
forall a. Maybe a
Nothing}

-- | Delete drained concurrency rows with no live job. Returns the number pruned.
pruneConcurrencyKeys
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
pruneConcurrencyKeys :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
pruneConcurrencyKeys = (Text -> [Text] -> m Int64) -> m Int64
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry Text -> [Text] -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
Ops.pruneConcurrencyKeys

-- | Recompute the concurrency counts from live jobs, repairing any trigger drift.
-- Returns the rows repaired.
reconcileConcurrencyCounts
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
reconcileConcurrencyCounts :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
reconcileConcurrencyCounts = (Text -> [Text] -> m Int64) -> m Int64
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry Text -> [Text] -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
Ops.reconcileConcurrencyCounts

-- | Rebuild the concurrency counts when a crash truncated the UNLOGGED table. Returns the
-- rows repaired.
reconcileConcurrencyCountsIfStale
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
reconcileConcurrencyCountsIfStale :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
reconcileConcurrencyCountsIfStale = (Text -> [Text] -> m Int64) -> m Int64
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry Text -> [Text] -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
Ops.reconcileConcurrencyCountsIfStale

-- | Reconcile then prune. Skipped when no concurrency key exists. Returns the rows
-- repaired plus the rows pruned.
reconcileAndPruneConcurrency
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
reconcileAndPruneConcurrency :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
reconcileAndPruneConcurrency = (Text -> [Text] -> m Int64) -> m Int64
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry Text -> [Text] -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m Int64
Ops.reconcileAndPruneConcurrency

-- | List filtered jobs, newest first.
listJobsFiltered
  :: forall payload m
   . (QueueOperation m payload)
  => [Ops.JobFilter]
  -- ^ Composable filters
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
listJobsFiltered :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobFilter] -> Int -> Int -> m [JobRead payload]
listJobsFiltered [JobFilter]
filters Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [JobRead payload]) -> m [JobRead payload])
-> (Text -> Text -> m [JobRead payload]) -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ \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)]
Ops.listJobsFiltered Text
schemaName Text
tableName [JobFilter]
filters Int
limit Int
offset 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)
Ops.typedRow

-- | Count filtered jobs.
countJobsFiltered
  :: forall payload m
   . (QueueOperation m payload)
  => [Ops.JobFilter]
  -- ^ Composable filters
  -> m Int64
countJobsFiltered :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobFilter] -> m Int64
countJobsFiltered [JobFilter]
filters =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
Ops.countJobsFiltered Text
schemaName Text
tableName [JobFilter]
filters

-- | List filtered DLQ jobs, most recently failed first.
listDLQFiltered
  :: forall payload m
   . (QueueOperation m payload)
  => [Ops.JobFilter]
  -- ^ Composable filters
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [DLQ.DLQJob payload]
listDLQFiltered :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobFilter] -> Int -> Int -> m [DLQJob payload]
listDLQFiltered [JobFilter]
filters Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [DLQJob payload]) -> m [DLQJob payload])
-> (Text -> Text -> m [DLQJob payload]) -> m [DLQJob payload]
forall a b. (a -> b) -> a -> b
$ \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)]
Ops.listDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters Int
limit Int
offset m [DLQJob (Stored payload)]
-> ([DLQJob (Stored payload)] -> m [DLQJob payload])
-> m [DLQJob payload]
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (DLQJob (Stored payload) -> m (DLQJob payload))
-> [DLQJob (Stored payload)] -> m [DLQJob 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 DLQJob (Stored payload) -> m (DLQJob payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
DLQJob (Stored payload) -> m (DLQJob payload)
Ops.typedDLQRow

-- | Count filtered DLQ jobs.
countDLQFiltered
  :: forall payload m
   . (QueueOperation m payload)
  => [Ops.JobFilter]
  -- ^ Composable filters
  -> m Int64
countDLQFiltered :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobFilter] -> m Int64
countDLQFiltered [JobFilter]
filters = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
Ops.countDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters

-- | List filtered archived jobs, most recently completed first.
listArchiveFiltered
  :: forall payload m
   . (QueueOperation m payload)
  => [Ops.JobFilter]
  -- ^ Composable filters
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [Archive.ArchiveJob payload]
listArchiveFiltered :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobFilter] -> Int -> Int -> m [ArchiveJob payload]
listArchiveFiltered [JobFilter]
filters Int
limit Int
offset = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [ArchiveJob payload])
 -> m [ArchiveJob payload])
-> (Text -> Text -> m [ArchiveJob payload])
-> m [ArchiveJob payload]
forall a b. (a -> b) -> a -> b
$ \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)]
Ops.listArchiveFiltered Text
schemaName Text
tableName [JobFilter]
filters Maybe ArchiveSortColumn
forall a. Maybe a
Nothing Maybe SortDir
forall a. Maybe a
Nothing Int
limit Int
offset m [ArchiveJob (Stored payload)]
-> ([ArchiveJob (Stored payload)] -> m [ArchiveJob payload])
-> m [ArchiveJob payload]
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (ArchiveJob (Stored payload) -> m (ArchiveJob payload))
-> [ArchiveJob (Stored payload)] -> m [ArchiveJob 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 ArchiveJob (Stored payload) -> m (ArchiveJob payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
ArchiveJob (Stored payload) -> m (ArchiveJob payload)
Ops.typedArchiveRow

-- | Count filtered archived jobs.
countArchiveFiltered
  :: forall payload m
   . (QueueOperation m payload)
  => [Ops.JobFilter]
  -- ^ Composable filters
  -> m Int64
countArchiveFiltered :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobFilter] -> m Int64
countArchiveFiltered [JobFilter]
filters =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
Ops.countArchiveFiltered Text
schemaName Text
tableName [JobFilter]
filters

-- | Move a job to the DLQ. Returns 0 for a job another worker holds.
moveToDLQ
  :: forall payload m
   . (MonadArbiter m)
  => Text
  -- ^ Error message (the final error that caused the DLQ move)
  -> JobRead payload
  -> m Int64
moveToDLQ :: forall payload (m :: * -> *).
MonadArbiter m =>
Text -> JobRead payload -> m Int64
moveToDLQ Text
errorMsg JobRead payload
job =
  JobRead payload -> (Text -> Text -> m Int64) -> m Int64
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    TreeLocks -> Text -> Text -> Text -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
TreeLocks -> Text -> Text -> Text -> JobRead payload -> m Int64
Ops.moveToDLQ TreeLocks
Ops.TakeLocks Text
schemaName Text
tableName Text
errorMsg JobRead payload
job

-- | Move a batch of jobs to the DLQ, skipping any another worker holds. Returns the
-- number moved.
moveToDLQBatch
  :: forall payload m
   . (MonadArbiter m)
  => [(JobRead payload, Text)]
  -- ^ List of (job, error message) pairs. All jobs must be from the same queue.
  -> m Int64
moveToDLQBatch :: forall payload (m :: * -> *).
MonadArbiter m =>
[(JobRead payload, Text)] -> m Int64
moveToDLQBatch [(JobRead payload, Text)]
jobsWithErrors =
  ((JobRead payload, Text) -> JobRead payload)
-> Int64
-> [(JobRead payload, Text)]
-> (Text -> Text -> m Int64)
-> m Int64
forall (m :: * -> *) item payload a.
MonadArbiter m =>
(item -> JobRead payload)
-> a -> [item] -> (Text -> Text -> m a) -> m a
onJobs (JobRead payload, Text) -> JobRead payload
forall a b. (a, b) -> a
fst Int64
0 [(JobRead payload, Text)]
jobsWithErrors ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [(JobRead payload, Text)] -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [(JobRead payload, Text)] -> m Int64
Ops.moveToDLQBatch Text
schemaName Text
tableName [(JobRead payload, Text)]
jobsWithErrors

-- | List DLQ jobs, most recently failed first.
listDLQJobs
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [DLQ.DLQJob payload]
listDLQJobs :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> m [DLQJob payload]
listDLQJobs Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [DLQJob payload]) -> m [DLQJob payload])
-> (Text -> Text -> m [DLQJob payload]) -> m [DLQJob payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int -> Int -> m [DLQJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [DLQJob (Stored payload)]
Ops.listDLQJobs Text
schemaName Text
tableName Int
limit Int
offset m [DLQJob (Stored payload)]
-> ([DLQJob (Stored payload)] -> m [DLQJob payload])
-> m [DLQJob payload]
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (DLQJob (Stored payload) -> m (DLQJob payload))
-> [DLQJob (Stored payload)] -> m [DLQJob 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 DLQJob (Stored payload) -> m (DLQJob payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
DLQJob (Stored payload) -> m (DLQJob payload)
Ops.typedDLQRow

-- | Retry the whole DLQ tree that holds a DLQ row, with a fresh attempt count. The dedup
-- key is left behind. Returns the restored job for that row. 'Nothing' when the row is
-- gone or the root's parent no longer exists.
retryFromDLQ
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ DLQ primary key ('Arbiter.Core.Job.DLQ.dlqPrimaryKey')
  -> m (Maybe (JobRead payload))
retryFromDLQ :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Maybe (JobRead payload))
retryFromDLQ Int64
dlqId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe (JobRead payload)))
 -> m (Maybe (JobRead payload)))
-> (Text -> Text -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
  m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
Ops.retryFromDLQ Text
schemaName Text
tableName Int64
dlqId m (Maybe (JobRead (Stored payload)))
-> (Maybe (JobRead (Stored 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
>>= (JobRead (Stored payload) -> m (JobRead payload))
-> Maybe (JobRead (Stored payload)) -> m (Maybe (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) -> Maybe a -> f (Maybe b)
traverse JobRead (Stored payload) -> m (JobRead payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
Ops.typedRow)

-- | 'retryFromDLQ' with a replacement payload. The columns the payload derives are
-- derived again from it.
retryFromDLQWithPayload
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ DLQ primary key ('Arbiter.Core.Job.DLQ.dlqPrimaryKey')
  -> payload
  -> m (Maybe (JobRead payload))
retryFromDLQWithPayload :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> payload -> m (Maybe (JobRead payload))
retryFromDLQWithPayload Int64
dlqId payload
replacement = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe (JobRead payload)))
 -> m (Maybe (JobRead payload)))
-> (Text -> Text -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
  m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction
    (Text
-> Text
-> Int64
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int64
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
Ops.retryFromDLQWithPayload Text
schemaName Text
tableName Int64
dlqId (payload -> Maybe payload
forall a. a -> Maybe a
Just payload
replacement) m (Maybe (JobRead (Stored payload)))
-> (Maybe (JobRead (Stored 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
>>= (JobRead (Stored payload) -> m (JobRead payload))
-> Maybe (JobRead (Stored payload)) -> m (Maybe (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) -> Maybe a -> f (Maybe b)
traverse JobRead (Stored payload) -> m (JobRead payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
Ops.typedRow)

-- | 'retryFromDLQ' over several DLQ rows, each tree in its own transaction. A row
-- restored with an earlier row's tree is skipped. Returns the number of rows whose
-- retry restored a tree.
retryFromDLQBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -- ^ DLQ primary keys ('Arbiter.Core.Job.DLQ.dlqPrimaryKey')
  -> m Int64
retryFromDLQBatch :: forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m Int64
retryFromDLQBatch [Int64]
dlqIds = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
Ops.retryFromDLQBatch Text
schemaName Text
tableName [Int64]
dlqIds

-- | 'retryFromDLQBatch' over every DLQ row the filters match when it starts, oldest
-- first. No filters retries the whole DLQ.
retryDLQFiltered
  :: forall payload m
   . (QueueOperation m payload)
  => [Ops.JobFilter]
  -- ^ Composable filters
  -> m Int64
retryDLQFiltered :: forall payload (m :: * -> *).
QueueOperation m payload =>
[JobFilter] -> m Int64
retryDLQFiltered [JobFilter]
filters = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
Ops.retryDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters

-- | Whether a DLQ row with the given DLQ primary key exists.
dlqJobExists
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ DLQ primary key ('Arbiter.Core.Job.DLQ.dlqPrimaryKey')
  -> m Bool
dlqJobExists :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Bool
dlqJobExists Int64
dlqId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Bool) -> m Bool)
-> (Text -> Text -> m Bool) -> m Bool
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Bool
Ops.dlqJobExists Text
schemaName Text
tableName Int64
dlqId

-- | Delete a DLQ job. Returns 0 when the row is already gone.
deleteDLQJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ DLQ primary key ('Arbiter.Core.Job.DLQ.dlqPrimaryKey')
  -> m Int64
deleteDLQJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
deleteDLQJob Int64
dlqId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.deleteDLQJob Text
schemaName Text
tableName Int64
dlqId

-- | Delete DLQ jobs by DLQ primary key. Returns the number deleted.
deleteDLQJobsBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -- ^ DLQ primary keys ('Arbiter.Core.Job.DLQ.dlqPrimaryKey')
  -> m Int64
deleteDLQJobsBatch :: forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m Int64
deleteDLQJobsBatch [Int64]
dlqIds =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
Ops.deleteDLQJobsBatch Text
schemaName Text
tableName [Int64]
dlqIds

-- | List archived jobs, most recently completed first.
listArchiveJobs
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [Archive.ArchiveJob payload]
listArchiveJobs :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> m [ArchiveJob payload]
listArchiveJobs Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [ArchiveJob payload])
 -> m [ArchiveJob payload])
-> (Text -> Text -> m [ArchiveJob payload])
-> m [ArchiveJob payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int -> Int -> m [ArchiveJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [ArchiveJob (Stored payload)]
Ops.listArchiveJobs Text
schemaName Text
tableName Int
limit Int
offset m [ArchiveJob (Stored payload)]
-> ([ArchiveJob (Stored payload)] -> m [ArchiveJob payload])
-> m [ArchiveJob payload]
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (ArchiveJob (Stored payload) -> m (ArchiveJob payload))
-> [ArchiveJob (Stored payload)] -> m [ArchiveJob 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 ArchiveJob (Stored payload) -> m (ArchiveJob payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
ArchiveJob (Stored payload) -> m (ArchiveJob payload)
Ops.typedArchiveRow

-- | Fetch a single archived job by its original job id.
getArchiveJobById
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Original job id
  -> m (Maybe (Archive.ArchiveJob payload))
getArchiveJobById :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Maybe (ArchiveJob payload))
getArchiveJobById Int64
jobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe (ArchiveJob payload)))
 -> m (Maybe (ArchiveJob payload)))
-> (Text -> Text -> m (Maybe (ArchiveJob payload)))
-> m (Maybe (ArchiveJob payload))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m (Maybe (ArchiveJob (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (ArchiveJob (Stored payload)))
Ops.getArchiveJobById Text
schemaName Text
tableName Int64
jobId m (Maybe (ArchiveJob (Stored payload)))
-> (Maybe (ArchiveJob (Stored payload))
    -> m (Maybe (ArchiveJob payload)))
-> m (Maybe (ArchiveJob payload))
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (ArchiveJob (Stored payload) -> m (ArchiveJob payload))
-> Maybe (ArchiveJob (Stored payload))
-> m (Maybe (ArchiveJob 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) -> Maybe a -> f (Maybe b)
traverse ArchiveJob (Stored payload) -> m (ArchiveJob payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
ArchiveJob (Stored payload) -> m (ArchiveJob payload)
Ops.typedArchiveRow

-- | List archived jobs in a group, most recently completed first, with pagination.
listArchiveJobsByGroup
  :: forall payload m
   . (QueueOperation m payload)
  => Text
  -- ^ Group key
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [Archive.ArchiveJob payload]
listArchiveJobsByGroup :: forall payload (m :: * -> *).
QueueOperation m payload =>
Text -> Int -> Int -> m [ArchiveJob payload]
listArchiveJobsByGroup Text
groupKey Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [ArchiveJob payload])
 -> m [ArchiveJob payload])
-> (Text -> Text -> m [ArchiveJob payload])
-> m [ArchiveJob payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text
-> Text -> Text -> Int -> Int -> m [ArchiveJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Text -> Int -> Int -> m [ArchiveJob (Stored payload)]
Ops.listArchiveJobsByGroup Text
schemaName Text
tableName Text
groupKey Int
limit Int
offset m [ArchiveJob (Stored payload)]
-> ([ArchiveJob (Stored payload)] -> m [ArchiveJob payload])
-> m [ArchiveJob payload]
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (ArchiveJob (Stored payload) -> m (ArchiveJob payload))
-> [ArchiveJob (Stored payload)] -> m [ArchiveJob 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 ArchiveJob (Stored payload) -> m (ArchiveJob payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
ArchiveJob (Stored payload) -> m (ArchiveJob payload)
Ops.typedArchiveRow

-- | Delete one archived job by its archive primary key. Returns rows deleted.
deleteArchiveJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Archive primary key
  -> m Int64
deleteArchiveJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
deleteArchiveJob Int64
archiveId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.deleteArchiveJob Text
schemaName Text
tableName Int64
archiveId

-- | Delete archived jobs by archive primary key. Returns rows deleted.
deleteArchiveJobsBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -- ^ Archive primary keys
  -> m Int64
deleteArchiveJobsBatch :: forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m Int64
deleteArchiveJobsBatch [Int64]
archiveIds =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
Ops.deleteArchiveJobsBatch Text
schemaName Text
tableName [Int64]
archiveIds

-- | Re-enqueue an archived job as a fresh standalone job, keeping the archive
-- row. Returns the new job, or @Nothing@ if the archive row no longer exists.
reEnqueueFromArchive
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Archive primary key
  -> m (Maybe (JobRead payload))
reEnqueueFromArchive :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Maybe (JobRead payload))
reEnqueueFromArchive Int64
archiveId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe (JobRead payload)))
 -> m (Maybe (JobRead payload)))
-> (Text -> Text -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
  m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
Ops.reEnqueueFromArchive Text
schemaName Text
tableName Int64
archiveId m (Maybe (JobRead (Stored payload)))
-> (Maybe (JobRead (Stored 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
>>= (JobRead (Stored payload) -> m (JobRead payload))
-> Maybe (JobRead (Stored payload)) -> m (Maybe (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) -> Maybe a -> f (Maybe b)
traverse JobRead (Stored payload) -> m (JobRead payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
Ops.typedRow)

-- | 'reEnqueueFromArchive' with a replacement payload. The columns the payload
-- derives are derived again from it.
reEnqueueFromArchiveWithPayload
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Archive primary key
  -> payload
  -> m (Maybe (JobRead payload))
reEnqueueFromArchiveWithPayload :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> payload -> m (Maybe (JobRead payload))
reEnqueueFromArchiveWithPayload Int64
archiveId payload
replacement = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe (JobRead payload)))
 -> m (Maybe (JobRead payload)))
-> (Text -> Text -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
  m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction
    (Text
-> Text
-> Int64
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> Int64
-> Maybe payload
-> m (Maybe (JobRead (Stored payload)))
Ops.reEnqueueFromArchiveWithPayload Text
schemaName Text
tableName Int64
archiveId (payload -> Maybe payload
forall a. a -> Maybe a
Just payload
replacement) m (Maybe (JobRead (Stored payload)))
-> (Maybe (JobRead (Stored 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
>>= (JobRead (Stored payload) -> m (JobRead payload))
-> Maybe (JobRead (Stored payload)) -> m (Maybe (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) -> Maybe a -> f (Maybe b)
traverse JobRead (Stored payload) -> m (JobRead payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
Ops.typedRow)

-- | List jobs, newest first.
listJobs
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
listJobs :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> m [JobRead payload]
listJobs Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [JobRead payload]) -> m [JobRead payload])
-> (Text -> Text -> m [JobRead payload]) -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int -> Int -> m [JobRead (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int -> Int -> m [JobRead (Stored payload)]
Ops.listJobs Text
schemaName Text
tableName Int
limit Int
offset 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)
Ops.typedRow

-- | Fetch a job by id.
getJobById
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> m (Maybe (JobRead payload))
getJobById :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Maybe (JobRead payload))
getJobById Int64
jobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe (JobRead payload)))
 -> m (Maybe (JobRead payload)))
-> (Text -> Text -> m (Maybe (JobRead payload)))
-> m (Maybe (JobRead payload))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
Ops.getJobById Text
schemaName Text
tableName Int64
jobId m (Maybe (JobRead (Stored payload)))
-> (Maybe (JobRead (Stored 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
>>= (JobRead (Stored payload) -> m (JobRead payload))
-> Maybe (JobRead (Stored payload)) -> m (Maybe (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) -> Maybe a -> f (Maybe b)
traverse JobRead (Stored payload) -> m (JobRead payload)
forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
Ops.typedRow

-- | Whether a job with the given id exists in this payload's queue table.
jobExists
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> m Bool
jobExists :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Bool
jobExists Int64
jobId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Bool) -> m Bool)
-> (Text -> Text -> m Bool) -> m Bool
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Bool
Ops.jobExists Text
schemaName Text
tableName Int64
jobId

-- | A page of a group's jobs, newest first.
listJobsByGroup
  :: forall payload m
   . (QueueOperation m payload)
  => Text
  -- ^ Group key
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
listJobsByGroup :: forall payload (m :: * -> *).
QueueOperation m payload =>
Text -> Int -> Int -> m [JobRead payload]
listJobsByGroup Text
groupKey Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [JobRead payload]) -> m [JobRead payload])
-> (Text -> Text -> m [JobRead payload]) -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Text -> Int -> Int -> m [JobRead (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Text -> Int -> Int -> m [JobRead (Stored payload)]
Ops.listJobsByGroup Text
schemaName Text
tableName Text
groupKey Int
limit Int
offset 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)
Ops.typedRow

-- | List a parent's children, newest first.
listJobsByParent
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent id
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
listJobsByParent :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> Int -> Int -> m [JobRead payload]
listJobsByParent Int64
pid Int
limit Int
offset =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m [JobRead payload]) -> m [JobRead payload])
-> (Text -> Text -> m [JobRead payload]) -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> Int -> Int -> m [JobRead (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> Int -> Int -> m [JobRead (Stored payload)]
Ops.listJobsByParent Text
schemaName Text
tableName Int64
pid Int
limit Int
offset 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)
Ops.typedRow

-- | Delete a job by id. Returns 0 for a job with children. 'cancelJobCascade' deletes
-- those.
cancelJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> m Int64
cancelJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
cancelJob Int64
jobId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.cancelJob Text
schemaName Text
tableName Int64
jobId

-- | Delete jobs by id. A job with children is skipped. Returns the number deleted.
cancelJobsBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -- ^ Job ids
  -> m Int64
cancelJobsBatch :: forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m Int64
cancelJobsBatch [Int64]
jobIds = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
Ops.cancelJobsBatch Text
schemaName Text
tableName [Int64]
jobIds

-- | Cascade-cancel a job subtree. Flags still-live claimed jobs, deletes the rest,
-- and NOTIFYs the queue's cancel channel for every claimed job affected. Returns the
-- number flagged or deleted.
forceCancelJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Root job id
  -> m Int64
forceCancelJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
forceCancelJob Int64
jobId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.forceCancelJob Text
schemaName Text
tableName Int64
jobId

-- | Make a delayed or retrying job immediately visible. Refuses an in-flight or suspended job.
promoteJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> m Int64
promoteJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
promoteJob Int64
jobId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.promoteJob Text
schemaName Text
tableName Int64
jobId

-- | Set when a job next becomes visible. Refuses an in-flight, suspended,
-- force-cancel-flagged or exhausted job.
rescheduleJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> UTCTime
  -- ^ When the job becomes visible
  -> m Int64
rescheduleJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> UTCTime -> m Int64
rescheduleJob Int64
jobId UTCTime
runAt =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> UTCTime -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> UTCTime -> m Int64
Ops.rescheduleJob Text
schemaName Text
tableName Int64
jobId UTCTime
runAt

-- | A queue's per-status counts and backlog ages.
getQueueStats
  :: forall payload m
   . (QueueOperation m payload)
  => m Ops.QueueStats
getQueueStats :: forall payload (m :: * -> *).
QueueOperation m payload =>
m QueueStats
getQueueStats = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m QueueStats) -> m QueueStats)
-> (Text -> Text -> m QueueStats) -> m QueueStats
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [Text] -> m QueueStats
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Text] -> m QueueStats
Ops.getQueueStats Text
schemaName Text
tableName (forall payload. HasKind payload => [Text]
kindsFor @payload)

-- | Count every job in the queue.
countJobs
  :: forall payload m
   . (QueueOperation m payload)
  => m Int64
countJobs :: forall payload (m :: * -> *). QueueOperation m payload => m Int64
countJobs = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
Ops.countJobs

-- | Count a group's jobs.
countJobsByGroup
  :: forall payload m
   . (QueueOperation m payload)
  => Text
  -- ^ Group key
  -> m Int64
countJobsByGroup :: forall payload (m :: * -> *).
QueueOperation m payload =>
Text -> m Int64
countJobsByGroup Text
groupKey =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Text -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> m Int64
Ops.countJobsByGroup Text
schemaName Text
tableName Text
groupKey

-- | Count a parent's children.
countJobsByParent
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent id
  -> m Int64
countJobsByParent :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
countJobsByParent Int64
pid = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.countJobsByParent Text
schemaName Text
tableName Int64
pid

-- | Count the queue's DLQ jobs.
countDLQJobs
  :: forall payload m
   . (QueueOperation m payload)
  => m Int64
countDLQJobs :: forall payload (m :: * -> *). QueueOperation m payload => m Int64
countDLQJobs = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
Ops.countDLQJobs

-- | Child counts as @(total, suspended)@ per parent id, over a batch. Parents with none
-- are absent.
countChildrenBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -- ^ Parent job ids
  -> m (Map Int64 (Int64, Int64))
countChildrenBatch :: forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m (Map Int64 (Int64, Int64))
countChildrenBatch [Int64]
ids = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Map Int64 (Int64, Int64)))
 -> m (Map Int64 (Int64, Int64)))
-> (Text -> Text -> m (Map Int64 (Int64, Int64)))
-> m (Map Int64 (Int64, Int64))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> [Int64] -> m (Map Int64 (Int64, Int64))
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m (Map Int64 (Int64, Int64))
Ops.countChildrenBatch Text
schemaName Text
tableName [Int64]
ids

-- | Count a parent's children in the DLQ.
countDLQChildren
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> m Int64
countDLQChildren :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
countDLQChildren Int64
parentJobId =
  Int64 -> Int64 -> Map Int64 Int64 -> Int64
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Int64
0 Int64
parentJobId (Map Int64 Int64 -> Int64) -> m (Map Int64 Int64) -> m Int64
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m (Map Int64 Int64)
countDLQChildrenBatch @payload [Int64
parentJobId]

-- | DLQ child counts per parent id, over a batch. Parents with none are absent.
countDLQChildrenBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -- ^ Parent job ids
  -> m (Map Int64 Int64)
countDLQChildrenBatch :: forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m (Map Int64 Int64)
countDLQChildrenBatch [Int64]
ids =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Map Int64 Int64)) -> m (Map Int64 Int64))
-> (Text -> Text -> m (Map Int64 Int64)) -> m (Map Int64 Int64)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> [Int64] -> m (Map Int64 Int64)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m (Map Int64 Int64)
Ops.countDLQChildrenBatch Text
schemaName Text
tableName [Int64]
ids

-- | Suspend every unsuspended visible job in a parent's subtree, lapsed leases included.
-- Jobs with a future visibility (delayed, backoff, throttled, leased) are skipped.
-- Returns the number suspended.
pauseChildren
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> m Int64
pauseChildren :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
pauseChildren Int64
parentJobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.pauseChildren Text
schemaName Text
tableName Int64
parentJobId

-- | Resume the suspended jobs in a parent's subtree. A finalizer with children still in
-- the queue stays suspended. Returns the number resumed.
resumeChildren
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> m Int64
resumeChildren :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
resumeChildren Int64
parentJobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.resumeChildren Text
schemaName Text
tableName Int64
parentJobId

-- | Delete a job and every descendant under it. Returns the number deleted.
cancelJobCascade
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Root job id
  -> m Int64
cancelJobCascade :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
cancelJobCascade Int64
jobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    [Int64] -> Int64
forall i a. Num i => [a] -> i
genericLength ([Int64] -> Int64) -> m [Int64] -> m 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]
Ops.cancelJobCascade Text
schemaName Text
tableName Int64
jobId

-- | Suspend a job, making it unclaimable. Refuses an in-flight job.
suspendJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> m Int64
suspendJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
suspendJob Int64
jobId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.suspendJob Text
schemaName Text
tableName Int64
jobId

-- | Resume a suspended job. Refuses a finalizer with children still in the main queue.
resumeJob
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> m Int64
resumeJob :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m Int64
resumeJob Int64
jobId = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.resumeJob Text
schemaName Text
tableName Int64
jobId

-- | Upsert a child's result, encoded as the queue's declared result type and keyed by
-- @(parent_id, child_id)@. A second write replaces the first. Its foreign key cascades.
-- Acking the parent clears it.
-- Returns 0 for a result that stores nothing.
insertResult
  :: forall payload m
   . (EncodeJobResult (ResultOf m payload), QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> Int64
  -- ^ Child job id
  -> ResultOf m payload
  -- ^ Result value
  -> m Int64
insertResult :: forall payload (m :: * -> *).
(EncodeJobResult (ResultOf m payload), QueueOperation m payload) =>
Int64 -> Int64 -> ResultOf m payload -> m Int64
insertResult Int64
parentJobId Int64
childId ResultOf m payload
result =
  m Int64 -> (Value -> m Int64) -> Maybe Value -> 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) (forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> Int64 -> Value -> m Int64
insertResultUnsafe @payload Int64
parentJobId Int64
childId) (SpecResult (MatchIn payload '[] (RegistryOf m)) -> Maybe Value
forall a. EncodeJobResult a => a -> Maybe Value
encodeJobResult ResultOf m payload
SpecResult (MatchIn payload '[] (RegistryOf m))
result)

-- | 'insertResult' with a raw JSON value, as an upsert, bypassing the queue's declared
-- result type. A value @Arbiter.Worker.childResults@ cannot decode surfaces there as a
-- 'Left'.
insertResultUnsafe
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> Int64
  -- ^ Child job id
  -> Value
  -- ^ Encoded result value
  -> m Int64
insertResultUnsafe :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> Int64 -> Value -> m Int64
insertResultUnsafe Int64
parentJobId Int64
childId Value
result =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> Int64 -> Value -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> Int64 -> Value -> m Int64
Ops.insertResult Text
schemaName Text
tableName Int64
parentJobId Int64
childId Value
result

-- | A parent's child results, keyed by child id.
getResultsByParent
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> m (Map Int64 Value)
getResultsByParent :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Map Int64 Value)
getResultsByParent Int64
parentJobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Map Int64 Value)) -> m (Map Int64 Value))
-> (Text -> Text -> m (Map Int64 Value)) -> m (Map Int64 Value)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m (Map Int64 Value)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Map Int64 Value)
Ops.getResultsByParent Text
schemaName Text
tableName Int64
parentJobId

-- | The last errors of a parent's children in the DLQ, keyed by child id. Children with
-- no recorded error are left out.
getDLQChildErrorsByParent
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> m (Map Int64 Text)
getDLQChildErrorsByParent :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Map Int64 Text)
getDLQChildErrorsByParent Int64
parentJobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Map Int64 Text)) -> m (Map Int64 Text))
-> (Text -> Text -> m (Map Int64 Text)) -> m (Map Int64 Text)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m (Map Int64 Text)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Map Int64 Text)
Ops.getDLQChildErrorsByParent Text
schemaName Text
tableName Int64
parentJobId

-- | Read all child data for a rollup finalizer. Returns
-- @(childId->result, childId->error, parentStateSnapshot, dlqRowId->error)@.
readChildResultsRaw
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent job id
  -> m (Map Int64 Value, Map Int64 Text, Maybe Value, Map Int64 Text)
readChildResultsRaw :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64
-> m (Map Int64 Value, Map Int64 Text, Maybe Value, Map Int64 Text)
readChildResultsRaw Int64
parentJobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text
  -> Text
  -> m (Map Int64 Value, Map Int64 Text, Maybe Value,
        Map Int64 Text))
 -> m (Map Int64 Value, Map Int64 Text, Maybe Value,
       Map Int64 Text))
-> (Text
    -> Text
    -> m (Map Int64 Value, Map Int64 Text, Maybe Value,
          Map Int64 Text))
-> m (Map Int64 Value, Map Int64 Text, Maybe Value, Map Int64 Text)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    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)
Ops.readChildResultsRaw Text
schemaName Text
tableName Int64
parentJobId

-- | Snapshot results into @parent_state@ before DLQ move.
persistParentState
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> Value
  -- ^ Snapshot to store
  -> m Int64
persistParentState :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> Value -> m Int64
persistParentState Int64
jobId Value
state =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m Int64) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> Value -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> Value -> m Int64
Ops.persistParentState Text
schemaName Text
tableName Int64
jobId Value
state

-- | Read raw @parent_state@ snapshot from the DB.
getParentStateSnapshot
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Job id
  -> m (Maybe Value)
getParentStateSnapshot :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Maybe Value)
getParentStateSnapshot Int64
jobId =
  forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Maybe Value)) -> m (Maybe Value))
-> (Text -> Text -> m (Maybe Value)) -> m (Maybe Value)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    Text -> Text -> Int64 -> m (Maybe Value)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe Value)
Ops.getParentStateSnapshot Text
schemaName Text
tableName Int64
jobId

-- | Refresh every registered queue's groups table to the end. Returns the rows rewritten
-- and the queue names that failed.
refreshAllGroupsFully
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m (Int64, [Text])
refreshAllGroupsFully :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m (Int64, [Text])
refreshAllGroupsFully = (Text -> [Text] -> m (Int64, [Text])) -> m (Int64, [Text])
forall (m :: * -> *) a.
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
(Text -> [Text] -> m a) -> m a
onRegistry Text -> [Text] -> m (Int64, [Text])
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> m (Int64, [Text])
Ops.refreshAllGroupsFully

-- | Register or refresh a worker and return its effective pause state (worker or
-- queue). 'Nothing' when the upsert returns no row.
registerWorker
  :: (MonadArbiter m)
  => UUID
  -- ^ Worker id
  -> Text
  -- ^ Queue name
  -> Maybe Text
  -- ^ Host name
  -> Maybe Int32
  -- ^ Worker thread count
  -> NominalDiffTime
  -- ^ Heartbeat age after which the worker counts as stale
  -> Maybe Value
  -- ^ Free-form metadata
  -> m (Maybe Bool)
registerWorker :: forall (m :: * -> *).
MonadArbiter m =>
UUID
-> Text
-> Maybe Text
-> Maybe Int32
-> NominalDiffTime
-> Maybe Value
-> m (Maybe Bool)
registerWorker UUID
workerId Text
queue Maybe Text
host Maybe Int32
threads NominalDiffTime
staleThreshold Maybe Value
metadata =
  (Text -> m (Maybe Bool)) -> m (Maybe Bool)
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m (Maybe Bool)) -> m (Maybe Bool))
-> (Text -> m (Maybe Bool)) -> m (Maybe Bool)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text
-> UUID
-> Text
-> Maybe Text
-> Maybe Int32
-> NominalDiffTime
-> Maybe Value
-> m (Maybe Bool)
forall (m :: * -> *).
MonadArbiter m =>
Text
-> UUID
-> Text
-> Maybe Text
-> Maybe Int32
-> NominalDiffTime
-> Maybe Value
-> m (Maybe Bool)
Ops.registerWorker Text
schemaName UUID
workerId Text
queue Maybe Text
host Maybe Int32
threads NominalDiffTime
staleThreshold Maybe Value
metadata

-- | Record a heartbeat and return the worker's effective pause state (worker or queue).
-- 'Nothing' when the worker has no registry row.
heartbeatWorker :: (MonadArbiter m) => UUID -> m (Maybe Bool)
heartbeatWorker :: forall (m :: * -> *). MonadArbiter m => UUID -> m (Maybe Bool)
heartbeatWorker UUID
workerId = (Text -> m (Maybe Bool)) -> m (Maybe Bool)
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m (Maybe Bool)) -> m (Maybe Bool))
-> (Text -> m (Maybe Bool)) -> m (Maybe Bool)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> UUID -> m (Maybe Bool)
forall (m :: * -> *).
MonadArbiter m =>
Text -> UUID -> m (Maybe Bool)
Ops.heartbeatWorker Text
schemaName UUID
workerId

-- | Set a worker's pause flag and notify the worker.
setWorkerPaused :: (MonadArbiter m) => UUID -> Bool -> m Int64
setWorkerPaused :: forall (m :: * -> *). MonadArbiter m => UUID -> Bool -> m Int64
setWorkerPaused UUID
workerId Bool
paused = (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> UUID -> Bool -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> UUID -> Bool -> m Int64
Ops.setWorkerPaused Text
schemaName UUID
workerId Bool
paused

-- | Mark a worker as gracefully draining.
markWorkerShuttingDown :: (MonadArbiter m) => UUID -> m Int64
markWorkerShuttingDown :: forall (m :: * -> *). MonadArbiter m => UUID -> m Int64
markWorkerShuttingDown UUID
workerId = (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> UUID -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> UUID -> m Int64
Ops.markWorkerShuttingDown Text
schemaName UUID
workerId

-- | Remove a worker registry row.
deregisterWorker :: (MonadArbiter m) => UUID -> m Int64
deregisterWorker :: forall (m :: * -> *). MonadArbiter m => UUID -> m Int64
deregisterWorker UUID
workerId = (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> UUID -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> UUID -> m Int64
Ops.deregisterWorker Text
schemaName UUID
workerId

-- | List workers, optionally filtered by queue and heartbeat age.
listWorkers :: (MonadArbiter m) => Maybe Text -> Maybe NominalDiffTime -> m [WorkerRow]
listWorkers :: forall (m :: * -> *).
MonadArbiter m =>
Maybe Text -> Maybe NominalDiffTime -> m [WorkerRow]
listWorkers Maybe Text
queue Maybe NominalDiffTime
liveSecs = (Text -> m [WorkerRow]) -> m [WorkerRow]
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m [WorkerRow]) -> m [WorkerRow])
-> (Text -> m [WorkerRow]) -> m [WorkerRow]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Maybe Text -> Maybe NominalDiffTime -> m [WorkerRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> Maybe NominalDiffTime -> m [WorkerRow]
Ops.listWorkers Text
schemaName Maybe Text
queue Maybe NominalDiffTime
liveSecs

-- | Delete workers whose last heartbeat is older than their stale threshold.
sweepStaleWorkers :: (MonadArbiter m) => m Int64
sweepStaleWorkers :: forall (m :: * -> *). MonadArbiter m => m Int64
sweepStaleWorkers = (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> m Int64
Ops.sweepStaleWorkers

-- | Ensure a queue's row exists.
ensureQueue :: (MonadArbiter m) => Text -> m Int64
ensureQueue :: forall (m :: * -> *). MonadArbiter m => Text -> m Int64
ensureQueue Text
queue = (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
Ops.ensureQueue Text
schemaName Text
queue

-- | Set a queue's pause flag and notify its workers.
setQueuePaused :: (MonadArbiter m) => Text -> Bool -> m Int64
setQueuePaused :: forall (m :: * -> *). MonadArbiter m => Text -> Bool -> m Int64
setQueuePaused Text
queue Bool
paused = (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> Bool -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Bool -> m Int64
Ops.setQueuePaused Text
schemaName Text
queue Bool
paused

-- | Set the age in seconds at which the reaper purges the queue's DLQ jobs, creating
-- the queue's row if missing. 'Nothing' keeps them. The retention must be positive.
-- Every pool and server sharing the schema applies it.
setQueueDLQRetention :: (MonadArbiter m) => Text -> Maybe NominalDiffTime -> m ()
setQueueDLQRetention :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe NominalDiffTime -> m ()
setQueueDLQRetention Text
queue Maybe NominalDiffTime
retention = (Text -> m ()) -> m ()
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m ()) -> m ()) -> (Text -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> Maybe NominalDiffTime -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe NominalDiffTime -> m ()
Ops.setQueueDLQRetention Text
schemaName Text
queue Maybe NominalDiffTime
retention

-- | Get one queue's row.
getQueue :: (MonadArbiter m) => Text -> m (Maybe QueueRow)
getQueue :: forall (m :: * -> *). MonadArbiter m => Text -> m (Maybe QueueRow)
getQueue Text
queue = (Text -> m (Maybe QueueRow)) -> m (Maybe QueueRow)
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m (Maybe QueueRow)) -> m (Maybe QueueRow))
-> (Text -> m (Maybe QueueRow)) -> m (Maybe QueueRow)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> m (Maybe QueueRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe QueueRow)
Ops.getQueue Text
schemaName Text
queue

-- | List the queue rows, ordered by queue name.
listQueues :: (MonadArbiter m) => m [QueueRow]
listQueues :: forall (m :: * -> *). MonadArbiter m => m [QueueRow]
listQueues = (Text -> m [QueueRow]) -> m [QueueRow]
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema Text -> m [QueueRow]
forall (m :: * -> *). MonadArbiter m => Text -> m [QueueRow]
Ops.listQueues

-- | List cron schedules ordered by name, optionally filtered by queue.
listCronSchedules
  :: forall m
   . (MonadArbiter m)
  => Maybe Text
  -- ^ Queue filter. 'Nothing' returns schedules for all queues.
  -> m [CronScheduleRow]
listCronSchedules :: forall (m :: * -> *).
MonadArbiter m =>
Maybe Text -> m [CronScheduleRow]
listCronSchedules Maybe Text
mQueue = (Text -> m [CronScheduleRow]) -> m [CronScheduleRow]
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m [CronScheduleRow]) -> m [CronScheduleRow])
-> (Text -> m [CronScheduleRow]) -> m [CronScheduleRow]
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Maybe Text -> m [CronScheduleRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> m [CronScheduleRow]
Ops.listCronSchedules Text
schemaName Maybe Text
mQueue

-- | Get a single cron schedule by name.
getCronScheduleByName
  :: forall m
   . (MonadArbiter m)
  => Text
  -- ^ Schedule name
  -> m (Maybe CronScheduleRow)
getCronScheduleByName :: forall (m :: * -> *).
MonadArbiter m =>
Text -> m (Maybe CronScheduleRow)
getCronScheduleByName Text
scheduleName = (Text -> m (Maybe CronScheduleRow)) -> m (Maybe CronScheduleRow)
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m (Maybe CronScheduleRow)) -> m (Maybe CronScheduleRow))
-> (Text -> m (Maybe CronScheduleRow)) -> m (Maybe CronScheduleRow)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.getCronScheduleByName Text
schemaName Text
scheduleName

-- | Patch a cron schedule, writing the overrides as given. Returns rows affected, 0 for
-- a name that is not there or an empty patch. @Arbiter.Worker.updateCronScheduleChecked@ rejects overrides
-- the scheduler cannot parse.
updateCronScheduleUnchecked
  :: forall m
   . (MonadArbiter m)
  => Text
  -- ^ Schedule name
  -> CronScheduleUpdate
  -> m Int64
updateCronScheduleUnchecked :: forall (m :: * -> *).
MonadArbiter m =>
Text -> CronScheduleUpdate -> m Int64
updateCronScheduleUnchecked Text
scheduleName CronScheduleUpdate
upd =
  (Text -> m Int64) -> m Int64
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m Int64) -> m Int64) -> (Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
    Text -> Text -> CronScheduleUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> CronScheduleUpdate -> m Int64
Ops.updateCronSchedule Text
schemaName Text
scheduleName CronScheduleUpdate
upd

-- | Run @work@ at most once per @interval@ across every worker pool sharing
-- the same schema. 'Nothing' when the gate ran too recently or another pool holds the task.
runGated
  :: forall m a
   . (MonadArbiter m)
  => Text
  -- ^ Task identifier
  -> NominalDiffTime
  -- ^ Minimum interval between runs, in seconds.
  -> m a
  -> m (Maybe a)
runGated :: forall (m :: * -> *) a.
MonadArbiter m =>
Text -> NominalDiffTime -> m a -> m (Maybe a)
runGated Text
task NominalDiffTime
interval m a
work = (Text -> m (Maybe a)) -> m (Maybe a)
forall (m :: * -> *) a. MonadArbiter m => (Text -> m a) -> m a
onSchema ((Text -> m (Maybe a)) -> m (Maybe a))
-> (Text -> m (Maybe a)) -> m (Maybe a)
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> NominalDiffTime -> m a -> m (Maybe a)
forall (m :: * -> *) a.
MonadArbiter m =>
Text -> Text -> NominalDiffTime -> m a -> m (Maybe a)
Ops.runGated Text
schemaName Text
task NominalDiffTime
interval m a
work

-- | Insert a t'Arbiter.Core.JobTree.JobTree' atomically. Returns all inserted jobs in pre-order.
-- @Left@ on a dedup conflict at any node, with nothing committed.
insertJobTree
  :: forall payload m
   . (QueueOperation m payload)
  => JT.JobTree payload
  -> m (Either Text (NonEmpty (JobRead payload)))
insertJobTree :: forall payload (m :: * -> *).
QueueOperation m payload =>
JobTree payload -> m (Either Text (NonEmpty (JobRead payload)))
insertJobTree JobTree payload
tree = forall payload (m :: * -> *) a.
(KnownSymbol (TableForPayload payload (RegistryOf m)),
 MonadArbiter m) =>
(Text -> Text -> m a) -> m a
onQueue @payload ((Text -> Text -> m (Either Text (NonEmpty (JobRead payload))))
 -> m (Either Text (NonEmpty (JobRead payload))))
-> (Text -> Text -> m (Either Text (NonEmpty (JobRead payload))))
-> m (Either Text (NonEmpty (JobRead payload)))
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text
-> Text
-> JobTree payload
-> m (Either Text (NonEmpty (JobRead payload)))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> JobTree payload
-> m (Either Text (NonEmpty (JobRead payload)))
JT.insertJobTree Text
schemaName Text
tableName JobTree payload
tree

-- | Insert children under a claimed job, making it a rollup finalizer. Ack it in the
-- same transaction. Throws 'Arbiter.Core.Exceptions.JobGoneException' when the claim is
-- gone, and 'Arbiter.Core.Exceptions.JobScopedFailure' with 'Arbiter.Core.Exceptions.Retryable'
-- on a dedup conflict or a second spawn in the same round.
spawnChildren
  :: forall payload m
   . (JobOperation m payload)
  => JobRead payload
  -> NonEmpty (JobWrite payload)
  -> m (NonEmpty (JobRead payload))
spawnChildren :: forall payload (m :: * -> *).
JobOperation m payload =>
JobRead payload
-> NonEmpty (JobWrite payload) -> m (NonEmpty (JobRead payload))
spawnChildren JobRead payload
job NonEmpty (JobWrite payload)
children =
  Text
-> [JobWrite payload]
-> m (NonEmpty (JobRead payload))
-> m (NonEmpty (JobRead payload))
forall payload (m :: * -> *) a.
(HasKind payload, MonadUnliftIO m) =>
Text -> [JobWrite payload] -> m a -> m a
withPublishSpan (JobRead payload -> Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> q
Job.queueName JobRead payload
job) (NonEmpty (JobWrite payload) -> [JobWrite payload]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty (JobWrite payload)
children) (m (NonEmpty (JobRead payload)) -> m (NonEmpty (JobRead payload)))
-> m (NonEmpty (JobRead payload)) -> m (NonEmpty (JobRead payload))
forall a b. (a -> b) -> a -> b
$
    JobRead payload
-> (Text -> Text -> m (NonEmpty (JobRead payload)))
-> m (NonEmpty (JobRead payload))
forall (m :: * -> *) payload a.
MonadArbiter m =>
JobRead payload -> (Text -> Text -> m a) -> m a
onJob JobRead payload
job ((Text -> Text -> m (NonEmpty (JobRead payload)))
 -> m (NonEmpty (JobRead payload)))
-> (Text -> Text -> m (NonEmpty (JobRead payload)))
-> m (NonEmpty (JobRead payload))
forall a b. (a -> b) -> a -> b
$
      \Text
schemaName Text
tableName -> Text
-> Text
-> JobRead payload
-> NonEmpty (JobWrite payload)
-> m (NonEmpty (JobRead payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text
-> Text
-> JobRead payload
-> NonEmpty (JobWrite payload)
-> m (NonEmpty (JobRead payload))
Ops.spawnChildren Text
schemaName Text
tableName JobRead payload
job NonEmpty (JobWrite payload)
children