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

-- | The job queue operations with their table names resolved from the payload type. A
-- payload reaches the queue its registry entry declares.
module Arbiter.Core.HighLevel
  ( -- * Constraint Aliases
    QueueOperation
  , JobOperation
  , queueTable
  , RegistryAdmissionPolicies

    -- * Job Operations
  , insertJob
  , insertJobsBatch
  , insertJobsBatch_
  , claimNextVisibleJobs
  , claimNextVisibleJobsAs
  , claimNextVisibleJobsBatched
  , mkClaimSql
  , addRateLimitTokens
  , pruneRateLimitBuckets
  , resetRateLimitBuckets
  , listRateLimitPolicies
  , getRateLimitPolicy
  , rateLimitPolicyExists
  , listRateLimitBuckets
  , listConcurrencyPolicies
  , getConcurrencyPolicy
  , listConcurrencyKeys
  , updateRateLimitPolicyOverrides
  , setRateLimit
  , clearRateLimit
  , updateConcurrencyPolicyOverrides
  , setConcurrencyLimit
  , clearConcurrencyLimit
  , pruneConcurrencyKeys
  , reconcileConcurrencyCounts
  , reconcileConcurrencyCountsIfStale
  , reconcileAndPruneConcurrency
  , ackJob
  , ackJobsBatch
  , updateJobForRetry
  , nackJob
  , nackJobsBatch
  , setVisibilityTimeout
  , setVisibilityTimeoutBatch
  , SetVisibilityResult (..)

    -- * Filtered Query Operations
  , Ops.JobFilter (..)
  , listJobsFiltered
  , countJobsFiltered
  , listDLQFiltered
  , countDLQFiltered

    -- * Dead Letter Queue Operations
  , moveToDLQ
  , moveToDLQBatch
  , listDLQJobs
  , listArchiveJobs
  , getArchivedJobById
  , listArchivedJobsByGroupKey
  , deleteArchiveJob
  , deleteArchiveJobsBatch
  , reEnqueueFromArchive
  , retryFromDLQ
  , dlqJobExists
  , deleteDLQJob
  , deleteDLQJobsBatch

    -- * Admin Operations
  , listJobs
  , getJobById
  , jobExists
  , getJobsByGroup
  , getJobsByParent
  , cancelJob
  , cancelJobsBatch
  , forceCancelJob
  , promoteJob
  , Ops.QueueStats (..)
  , getQueueStats

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

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

    -- * Suspend/Resume Operations
  , suspendJob
  , resumeJob

    -- * Results Table Operations
  , insertResult
  , insertResultUnsafe
  , getResultsByParent
  , getDLQChildErrorsByParent
  , readChildResultsRaw
  , Ops.mergeRawChildResults
  , persistParentState
  , getParentStateSnapshot

    -- * Groups Table Operations
  , refreshAllGroupsFully

    -- * Worker Registry Operations
  , registerWorker
  , heartbeatWorker
  , setWorkerPaused
  , markWorkerShuttingDown
  , deregisterWorker
  , listWorkers
  , sweepStaleWorkers
  , WorkerRow (..)

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

    -- * Cron Schedule Operations
  , listCronSchedules
  , getCronScheduleByName
  , updateCronScheduleUnchecked

    -- * Global Gate
  , runGated

    -- * Job Tree DSL
  , insertJobTree

    -- * Re-exports
  , getSchema
  ) where

import Control.Monad (void, when)
import Data.Aeson (Value)
import Data.Int (Int32, Int64)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Maybe (fromMaybe)
import Data.Proxy (Proxy (..))
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (NominalDiffTime)
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.HighLevel.Runtime
  ( deregisterWorker
  , ensureQueue
  , getQueue
  , heartbeatWorker
  , listQueues
  , listWorkers
  , markWorkerShuttingDown
  , registerWorker
  , setQueuePaused
  , setWorkerPaused
  , sweepStaleWorkers
  )
import Arbiter.Core.Job.Archive qualified as Archive
import Arbiter.Core.Job.DLQ qualified as DLQ
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 (Policy (..), RateLimitKey (..))
import Arbiter.Core.RateLimit.Stats (RateLimitBucketView, RateLimitPolicyUpdate (..), RateLimitPolicyView)
import Arbiter.Core.Trace (withPublishSpan)
import Arbiter.Core.Worker (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)
  => (Text -> Text -> 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)

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 inserted job, or @Nothing@ if skipped by dedup
-- ('Arbiter.Core.Job.Dedup.IgnoreDuplicate').
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 that were inserted.
-- Dedup'd jobs are excluded. 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 count of inserted rows.
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 pool caps this path as it
-- caps any other claim.
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

-- | Add tokens to a key's bucket, capped at max, and wake any of its jobs parked
-- mid-wait. A no-op without a policy.
addRateLimitTokens
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => RateLimitKey
  -> Double
  -> m ()
addRateLimitTokens :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
RateLimitKey -> Double -> m ()
addRateLimitTokens RateLimitKey
key Double
amount = m () -> m ()
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.addRateLimitTokens schemaName key amount
  let queues = 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))
  void $ Ops.wakeThrottledJobsForKey schemaName queues key

-- | Override a policy with this shape, until it is cleared. Returns rows affected.
setRateLimit :: (MonadArbiter m, RegistryTables (RegistryOf m)) => Policy -> m Int64
setRateLimit :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Policy -> m Int64
setRateLimit Policy
policy =
  Text -> RateLimitPolicyUpdate -> m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides (Policy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf Policy
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 (Policy -> Double
policyMax Policy
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 (Policy -> Double
policyRefill Policy
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 (Policy -> NominalDiffTime
policyInterval Policy
policy)))
      }

-- | Drop a policy's overrides. Its declared params apply again. Returns rows affected.
clearRateLimit :: (MonadArbiter m, RegistryTables (RegistryOf m)) => Policy -> m Int64
clearRateLimit :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Policy -> m Int64
clearRateLimit Policy
policy =
  Text -> RateLimitPolicyUpdate -> m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> RateLimitPolicyUpdate -> m Int64
updateRateLimitPolicyOverrides (Policy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf Policy
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
      }

-- | Delete reclaimable idle (full) buckets. Returns the number pruned. The worker
-- 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.pruneRateLimitBuckets schemaName 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
$ do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  resetCount <- Ops.resetRateLimitBuckets schemaName prefix
  let queues = 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))
  _ <- Ops.wakeThrottledJobs schemaName queues prefix
  pure resetCount

-- | 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.listRateLimitPolicies schemaName (registryTableNames (Proxy @(RegistryOf m)))

-- | 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.getRateLimitPolicy schemaName (registryTableNames (Proxy @(RegistryOf m))) 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.rateLimitPolicyExists schemaName prefix

-- | List a prefix's buckets with fill levels, paginated.
listRateLimitBuckets
  :: forall m
   . (MonadArbiter m)
  => Text
  -> Int
  -> Int
  -> m [RateLimitBucketView]
listRateLimitBuckets :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Int -> Int -> m [RateLimitBucketView]
listRateLimitBuckets Text
prefix Int
limit Int
offset = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.listRateLimitBuckets schemaName prefix limit offset

-- | Set or clear a policy's override params and wake the prefix's parked 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  affected <- Ops.updateRateLimitPolicyOverrides schemaName prefix upd
  -- Wake after the override commits.
  when (affected > 0) $ do
    let queues = 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))
    void $ Ops.wakeThrottledJobs schemaName queues prefix
  pure affected

-- | List every concurrency pool 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.listConcurrencyPolicies schemaName

-- | One prefix's concurrency pool with its default/override limit and live aggregates.
-- 'Nothing' when the prefix has no pool.
getConcurrencyPolicy
  :: forall m
   . (MonadArbiter m)
  => Text
  -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy :: forall (m :: * -> *).
MonadArbiter m =>
Text -> m (Maybe ConcurrencyPolicyView)
getConcurrencyPolicy Text
prefix = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.getConcurrencyPolicy schemaName prefix

-- | List a prefix's keys with effective cap and in-flight fill fraction, paginated.
listConcurrencyKeys
  :: forall m
   . (MonadArbiter m)
  => Text
  -> Int
  -> Int
  -> m [ConcurrencyKeyView]
listConcurrencyKeys :: forall (m :: * -> *).
MonadArbiter m =>
Text -> Int -> Int -> m [ConcurrencyKeyView]
listConcurrencyKeys Text
prefix Int
limit Int
offset = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.listConcurrencyKeys schemaName prefix limit offset

-- | Apply a pool's override-limit patch on its policy row, retuning every key under
-- the prefix live. Lowering it does not preempt in-flight jobs until they drain.
-- 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.updateConcurrencyPolicyOverrides schemaName prefix upd

-- | Override a declared pool's limit for every key under it. Returns rows affected.
setConcurrencyLimit :: (MonadArbiter m) => ConcurrencyPolicy -> Int32 -> m Int64
setConcurrencyLimit :: forall (m :: * -> *).
MonadArbiter m =>
ConcurrencyPolicy -> Int32 -> m Int64
setConcurrencyLimit ConcurrencyPolicy
pool Int32
limit =
  Text -> ConcurrencyPolicyUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides (ConcurrencyPolicy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf ConcurrencyPolicy
pool) (Maybe (Maybe Int32) -> ConcurrencyPolicyUpdate
ConcurrencyPolicyUpdate (Maybe Int32 -> Maybe (Maybe Int32)
forall a. a -> Maybe a
Just (Int32 -> Maybe Int32
forall a. a -> Maybe a
Just Int32
limit)))

-- | Drop a pool'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
pool =
  Text -> ConcurrencyPolicyUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> ConcurrencyPolicyUpdate -> m Int64
updateConcurrencyPolicyOverrides (ConcurrencyPolicy -> Text
forall p. AdmissionPolicy p => p -> Text
policyPrefixOf ConcurrencyPolicy
pool) (Maybe (Maybe Int32) -> ConcurrencyPolicyUpdate
ConcurrencyPolicyUpdate (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. The reaper runs this.
pruneConcurrencyKeys
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
pruneConcurrencyKeys :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
pruneConcurrencyKeys = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.pruneConcurrencyKeys schemaName (registryTableNames (Proxy @(RegistryOf m)))

-- | Recompute the concurrency counts from live jobs, repairing any trigger drift.
reconcileConcurrencyCounts
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
reconcileConcurrencyCounts :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
reconcileConcurrencyCounts = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.reconcileConcurrencyCounts schemaName (registryTableNames (Proxy @(RegistryOf m)))

-- | Rebuild the concurrency counts when a crash truncated the UNLOGGED table. The
-- reaper runs this periodically.
reconcileConcurrencyCountsIfStale
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
reconcileConcurrencyCountsIfStale :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
reconcileConcurrencyCountsIfStale = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.reconcileConcurrencyCountsIfStale schemaName (registryTableNames (Proxy @(RegistryOf m)))

-- | Reconcile then prune. The reaper runs this.
reconcileAndPruneConcurrency
  :: forall m
   . (MonadArbiter m, RegistryTables (RegistryOf m))
  => m Int64
reconcileAndPruneConcurrency :: forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
reconcileAndPruneConcurrency = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.reconcileAndPruneConcurrency schemaName (registryTableNames (Proxy @(RegistryOf m)))

-- | Assemble a pool's claim statements once (see 'Ops.mkClaimSql'). Batch size 1
-- is the single-job claim.
mkClaimSql
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -> Int
  -> NominalDiffTime
  -> UUID
  -> m Ops.ClaimSql
mkClaimSql :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> NominalDiffTime -> UUID -> m ClaimSql
mkClaimSql Int
batchSize Int
poolSize 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 ClaimSql) -> m ClaimSql)
-> (Text -> Text -> m ClaimSql) -> m ClaimSql
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName ->
    ClaimSql -> m ClaimSql
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ClaimSql -> m ClaimSql) -> ClaimSql -> m ClaimSql
forall a b. (a -> b) -> a -> b
$ Proxy payload
-> Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> ClaimSql
forall payload (proxy :: * -> *).
JobPayload payload =>
proxy payload
-> Text
-> Text
-> Int
-> Int
-> NominalDiffTime
-> UUID
-> ClaimSql
Ops.mkClaimSql (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @payload) Text
schemaName Text
tableName Int
batchSize Int
poolSize NominalDiffTime
timeout UUID
workerId

-- | 'claimNextVisibleJobs' under a given worker id. The dispatcher claims this way.
claimNextVisibleJobsAs
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -> NominalDiffTime
  -> 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.
claimNextVisibleJobsBatched
  :: forall payload m
   . (QueueOperation m payload)
  => Int
  -- ^ Batch size: maximum number of jobs to claim per group.
  -> Int
  -- ^ Max groups: maximum number of groups/batches to claim.
  -> 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
maxGroups 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
maxGroups NominalDiffTime
timeout

-- | Ack a completed job, deleting it or suspending it when its children are still
-- running. Returns 1, or 0 for a job already gone.
ackJob
  :: forall payload m
   . (JobOperation m payload)
  => JobRead payload
  -> m Int64
ackJob :: forall payload (m :: * -> *).
JobOperation m payload =>
JobRead payload -> m Int64
ackJob JobRead payload
job = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
job
  Ops.ackJob schemaName tableName job

-- | 'ackJob' over a batch from one queue in one statement, returning the ids acked.
-- Reclaimed jobs are absent.
ackJobsBatch
  :: forall payload m
   . (JobOperation m payload)
  => [JobRead payload]
  -> m [Int64]
ackJobsBatch :: forall payload (m :: * -> *).
JobOperation m payload =>
[JobRead payload] -> m [Int64]
ackJobsBatch [] = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
ackJobsBatch jobs :: [JobRead payload]
jobs@(JobRead payload
firstJob : [JobRead payload]
_) = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
firstJob
  Ops.ackJobsBatch schemaName tableName jobs

-- | Park a failed job for its retry backoff. Returns 0 for a job another worker holds.
updateJobForRetry
  :: forall payload m
   . (JobOperation m payload)
  => NominalDiffTime
  -- ^ The delay before this job becomes visible again for retry.
  -> Text
  -- ^ An error message to store with the job.
  -> JobRead payload
  -> m Int64
updateJobForRetry :: forall payload (m :: * -> *).
JobOperation m payload =>
NominalDiffTime -> Text -> JobRead payload -> m Int64
updateJobForRetry NominalDiffTime
delay Text
errorMsg JobRead payload
job = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
job
  Ops.updateJobForRetry schemaName tableName delay errorMsg 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.
nackJob
  :: forall payload m
   . (JobOperation m payload)
  => JobRead payload
  -> m Int64
nackJob :: forall payload (m :: * -> *).
JobOperation m payload =>
JobRead payload -> m Int64
nackJob JobRead payload
job = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
job
  Ops.nackJob schemaName tableName job

-- | 'nackJob' over a batch from one queue in one statement, returning the ids nacked.
-- Jobs another worker holds are absent.
nackJobsBatch
  :: forall payload m
   . (JobOperation m payload)
  => [JobRead payload]
  -> m [Int64]
nackJobsBatch :: forall payload (m :: * -> *).
JobOperation m payload =>
[JobRead payload] -> m [Int64]
nackJobsBatch [] = [Int64] -> m [Int64]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
nackJobsBatch jobs :: [JobRead payload]
jobs@(JobRead payload
firstJob : [JobRead payload]
_) = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.nackJobsBatch schemaName (Job.queueName firstJob) jobs

-- | Extend a job's visibility timeout by hand, for a long-running job. Returns 0 for a
-- job that is gone, reclaimed or suspended. 'setVisibilityTimeoutBatch' tells them
-- apart.
setVisibilityTimeout
  :: forall payload m
   . (JobOperation m payload)
  => NominalDiffTime
  -- ^ The new visibility timeout (in seconds) from the current time.
  -> JobRead payload
  -> m Int64
setVisibilityTimeout :: forall payload (m :: * -> *).
JobOperation m payload =>
NominalDiffTime -> JobRead payload -> m Int64
setVisibilityTimeout NominalDiffTime
timeout JobRead payload
job = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
job
  Ops.setVisibilityTimeout schemaName tableName timeout job

-- | 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 a finalizer waiting on its children. It holds no lease.
    JobSuspended JobId
  | -- | The row changed under this claim mid-statement. Nothing was extended.
    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)

-- | 'setVisibilityTimeout' over a batch from one queue.
setVisibilityTimeoutBatch
  :: forall payload m
   . (JobOperation m payload)
  => 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 :: * -> *).
JobOperation m payload =>
NominalDiffTime -> [JobRead payload] -> m [SetVisibilityResult]
setVisibilityTimeoutBatch NominalDiffTime
_ [] = [SetVisibilityResult] -> m [SetVisibilityResult]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
setVisibilityTimeoutBatch NominalDiffTime
timeout jobs :: [JobRead payload]
jobs@(JobRead payload
firstJob : [JobRead payload]
_) = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
firstJob
  infos <- Ops.setVisibilityTimeoutBatch schemaName tableName timeout jobs
  let 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.
JobRecord payload key q insertedAt adm -> key
primaryKey JobRead payload
job, JobRead payload
job) | JobRead payload
job <- [JobRead payload]
jobs]
      toResult (Ops.VisibilityUpdateInfo Int64
jobId Bool
heartbeated Maybe Int64
mActual Bool
cancelled Bool
suspended 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.
JobRecord payload Int64 q insertedAt adm -> Int64
claimSeq Maybe (JobRead payload)
mJob
            heldHere :: Bool
heldHere = Bool
-> (JobRead payload -> Bool) -> Maybe (JobRead payload) -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
False (\JobRead payload
job -> JobRead payload -> Maybe UUID
forall payload q insertedAt adm.
JobRecord payload Int64 q insertedAt adm -> Maybe UUID
claimedBy JobRead payload
job Maybe UUID -> Maybe UUID -> Bool
forall a. Eq a => a -> a -> Bool
== Maybe UUID
holder) 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
1 -> 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
  pure $ map toResult infos

-- | Move a job to the DLQ. Returns 0 for a job another worker holds.
moveToDLQ
  :: forall payload m
   . (JobOperation m payload)
  => Text
  -- ^ Error message (the final error that caused the DLQ move)
  -> JobRead payload
  -> m Int64
moveToDLQ :: forall payload (m :: * -> *).
JobOperation m payload =>
Text -> JobRead payload -> m Int64
moveToDLQ Text
errorMsg JobRead payload
job = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
job
  Ops.moveToDLQ Ops.TakeLocks schemaName tableName errorMsg job

-- | 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> Int -> m [DLQJob payload]
Ops.listDLQJobs Text
schemaName Text
tableName Int
limit Int
offset

-- | 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> Int -> m [ArchiveJob payload]
Ops.listArchiveJobs Text
schemaName Text
tableName Int
limit Int
offset

-- | Fetch a single archived job by its original job id.
getArchivedJobById
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Original job id
  -> m (Maybe (Archive.ArchiveJob payload))
getArchivedJobById :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> m (Maybe (ArchiveJob payload))
getArchivedJobById 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 payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (ArchiveJob payload))
Ops.getArchivedJobById Text
schemaName Text
tableName Int64
jobId

-- | List archived jobs in a group, most recent first, with pagination.
listArchivedJobsByGroupKey
  :: forall payload m
   . (QueueOperation m payload)
  => Text
  -- ^ Group key
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [Archive.ArchiveJob payload]
listArchivedJobsByGroupKey :: forall payload (m :: * -> *).
QueueOperation m payload =>
Text -> Int -> Int -> m [ArchiveJob payload]
listArchivedJobsByGroupKey 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Text -> Int -> Int -> m [ArchiveJob payload]
Ops.listArchivedJobsByGroupKey Text
schemaName Text
tableName Text
groupKey Int
limit Int
offset

-- | 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 -> Text -> Text -> Int64 -> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (JobRead payload))
Ops.reEnqueueFromArchive Text
schemaName Text
tableName Int64
archiveId

-- | Retry a DLQ job, re-inserting it into the queue with a fresh attempt count.
-- 'Nothing' when the DLQ row is gone.
retryFromDLQ
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ DLQ job id
  -> 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 -> Text -> Text -> Int64 -> m (Maybe (JobRead payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (JobRead payload))
Ops.retryFromDLQ Text
schemaName Text
tableName Int64
dlqId

-- | Whether a DLQ job with the given id exists.
dlqJobExists
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -> 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 job id
  -> 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

-- | Move a batch of jobs to the DLQ, skipping any another worker holds. Returns the
-- number moved.
moveToDLQBatch
  :: forall payload m
   . (JobOperation m payload)
  => [(JobRead payload, Text)]
  -- ^ List of (job, error message) pairs. All jobs must be from the same queue.
  -> m Int64
moveToDLQBatch :: forall payload (m :: * -> *).
JobOperation m payload =>
[(JobRead payload, Text)] -> m Int64
moveToDLQBatch [] = Int64 -> m Int64
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0
moveToDLQBatch jobsWithErrors :: [(JobRead payload, Text)]
jobsWithErrors@((JobRead payload
firstJob, Text
_) : [(JobRead payload, Text)]
_) = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  let tableName = JobRead payload -> Text
forall payload key q insertedAt adm.
JobRecord payload key q insertedAt adm -> q
Job.queueName JobRead payload
firstJob
  Ops.moveToDLQBatch schemaName tableName jobsWithErrors

-- | Delete DLQ jobs by id. Returns the number deleted.
deleteDLQJobsBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -- ^ DLQ job ids
  -> 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

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

-- | 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [JobRead payload]
Ops.listJobsFiltered Text
schemaName Text
tableName [JobFilter]
filters Int
limit Int
offset

-- | 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> [JobFilter] -> Int -> Int -> m [DLQJob payload]
Ops.listDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters Int
limit Int
offset

-- | 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

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

-- | 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int -> Int -> m [JobRead payload]
Ops.listJobs Text
schemaName Text
tableName Int
limit Int
offset

-- | 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 payload))
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> m (Maybe (JobRead payload))
Ops.getJobById Text
schemaName Text
tableName Int64
jobId

-- | 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

-- | List a group's jobs.
getJobsByGroup
  :: forall payload m
   . (QueueOperation m payload)
  => Text
  -- ^ Group key
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
getJobsByGroup :: forall payload (m :: * -> *).
QueueOperation m payload =>
Text -> Int -> Int -> m [JobRead payload]
getJobsByGroup 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Text -> Int -> Int -> m [JobRead payload]
Ops.getJobsByGroup Text
schemaName Text
tableName Text
groupKey Int
limit Int
offset

-- | List a parent's children.
getJobsByParent
  :: forall payload m
   . (QueueOperation m payload)
  => Int64
  -- ^ Parent id
  -> Int
  -- ^ Limit
  -> Int
  -- ^ Offset
  -> m [JobRead payload]
getJobsByParent :: forall payload (m :: * -> *).
QueueOperation m payload =>
Int64 -> Int -> Int -> m [JobRead payload]
getJobsByParent 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 payload]
forall (m :: * -> *) payload.
(JobPayload payload, MonadArbiter m) =>
Text -> Text -> Int64 -> Int -> Int -> m [JobRead payload]
Ops.getJobsByParent Text
schemaName Text
tableName Int64
pid Int
limit Int
offset

-- | 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 a job and its descendants, interrupting any handler running one of them by
-- NOTIFY on the cancel channel.
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

-- | Delete jobs by id. 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

-- | Make a delayed or retrying job immediately visible. Refuses an in-flight 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

-- | 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 Operations
-- ---------------------------------------------------------------------------

-- | 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) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
Ops.countJobs Text
schemaName Text
tableName

-- | 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) -> m Int64)
-> (Text -> Text -> m Int64) -> m Int64
forall a b. (a -> b) -> a -> b
$ \Text
schemaName Text
tableName -> Text -> Text -> m Int64
forall (m :: * -> *). MonadArbiter m => Text -> Text -> m Int64
Ops.countDLQJobs Text
schemaName Text
tableName

-- | Child counts as @(total, paused)@ per parent id, over a batch. Parents with none
-- are absent.
countChildrenBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -> 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 DLQ'd children.
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 = do
  counts <- forall payload (m :: * -> *).
QueueOperation m payload =>
[Int64] -> m (Map Int64 Int64)
countDLQChildrenBatch @payload [Int64
parentJobId]
  pure $ fromMaybe 0 (Map.lookup parentJobId counts)

-- | DLQ child counts per parent id, over a batch. Parents with none are absent.
countDLQChildrenBatch
  :: forall payload m
   . (QueueOperation m payload)
  => [Int64]
  -> 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

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

-- | Suspend a parent's claimable children. In-flight ones are left alone. 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 a parent's suspended children. 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 -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.cancelJobCascade Text
schemaName Text
tableName Int64
jobId

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

-- | 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, making it claimable again.
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

-- ---------------------------------------------------------------------------
-- Results Table Operations
-- ---------------------------------------------------------------------------

-- | Insert a child's result, encoded as the queue's declared result type and keyed by
-- @(parent_id, child_id)@. 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, 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

-- | A parent's DLQ'd children's last errors, keyed by child id.
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
  -> Value
  -> 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
  -> 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

-- ---------------------------------------------------------------------------
-- Groups Table Operations
-- ---------------------------------------------------------------------------

-- | Schema-wide groups-table refresh, correcting every registered queue's summary drift.
-- Walks each groups table to the end, one bounded batch and one transaction at a time.
-- This is a repair operation. The reaper runs 'Ops.refreshAllGroups' for
-- a single batch per tick. 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.refreshAllGroupsFully schemaName (registryTableNames (Proxy @(RegistryOf m)))

-- ---------------------------------------------------------------------------
-- Cron Schedules
-- ---------------------------------------------------------------------------

-- | 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.listCronSchedules schemaName 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.getCronScheduleByName schemaName scheduleName

-- | Patch a cron schedule, writing the overrides as given. Returns rows affected, 0 for
-- a name that is not there. @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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.updateCronSchedule schemaName scheduleName upd

-- ---------------------------------------------------------------------------
-- Global Gate
-- ---------------------------------------------------------------------------

-- | Run @work@ at most once per @interval@ across every worker pool sharing
-- the same schema. See 'Ops.runGated'.
runGated
  :: forall m a
   . (MonadArbiter m)
  => Text
  -- ^ Task identifier
  -> NominalDiffTime
  -> 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 = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  Ops.runGated schemaName task interval work

-- ---------------------------------------------------------------------------
-- Job Tree DSL
-- ---------------------------------------------------------------------------

-- | Insert a 'JT.JobTree' atomically. Returns all inserted jobs (pre-order),
-- or @Left@ if the root has a dedup conflict. Rolls back on any failure.
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