{-# LANGUAGE AllowAmbiguousTypes #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE UndecidableInstances #-}

-- | REST API server for the Arbiter job queue, over any 'MonadArbiter' backend.
--
-- __Security:__ No built-in authentication. All endpoints are publicly
-- accessible. Add auth middleware before exposing to untrusted networks.
module Arbiter.Servant.Server
  ( -- * Server
    arbiterServer
  , arbiterServerHoisted
  , arbiterApp
  , runArbiterAPI
  , ArbiterServerConfig (..)
  , createArbiterServer
  , defaultStatsCacheTtl
  , defaultMaintenanceInterval
  , defaultMaintenanceBucketIdle
  , defaultMaintenanceSparseInterval
  , defaultMaintenanceTimeout
  , BuildServer (..)
  , CacheCell
  ) where

import Arbiter.Core.CronSchedule qualified as CS
import Arbiter.Core.Health qualified as Health
import Arbiter.Core.HighLevel qualified as HL
import Arbiter.Core.Job.Schema qualified as Schema
import Arbiter.Core.Job.Types (DedupKey (..), JobPayload, JobStatus, isRollup, kindsFor)
import Arbiter.Core.Job.Types qualified as Job
import Arbiter.Core.JobResult (EncodeJobResult, encodeJobResult)
import Arbiter.Core.Listen (Notification (..), withChannels)
import Arbiter.Core.MonadArbiter (HasRegistry, MonadArbiter, getListener, getSchema, withDbTransaction)
import Arbiter.Core.Operations qualified as Ops
import Arbiter.Core.QueueRegistry (JobPayloadRegistry, RegistryTables (..), SpecName, SpecPayload, SpecResult)
import Arbiter.Core.Queues qualified as Queues
import Arbiter.Core.RateLimit.Spec (RateLimitKey (..))
import Arbiter.Core.Sql.Jobs (ArchiveSortColumn, DLQSortColumn, JobFilter (..), JobSortColumn, SortDir)
import Arbiter.Core.Trace (withPublishSpan)
import Arbiter.Worker (MaintenancePace (..), runMaintenancePass, storeEncodedResult)
import Arbiter.Worker.Config
  ( defaultMaintenanceBucketIdle
  , defaultMaintenanceSparseInterval
  , defaultMaintenanceTimeout
  , maintenanceOpName
  )
import Arbiter.Worker.Cron (nextRunFromExpression, updateCronScheduleChecked)
import Arbiter.Worker.Logger
  ( FailureGates
  , LogConfig
  , LogLevel (..)
  , defaultLogConfig
  , newFailureGates
  , toHubLog
  , tryReportedOn
  )
import Control.Concurrent.STM
  ( TVar
  , atomically
  , check
  , modifyTVar'
  , newTChanIO
  , newTVarIO
  , readTChan
  , readTVar
  , readTVarIO
  , writeTChan
  )
import Control.Exception (bracket_)
import Control.Monad (guard, join, mfilter, void, when)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Data.Aeson (encode)
import Data.ByteString (ByteString)
import Data.ByteString.Builder qualified as Builder
import Data.ByteString.Lazy qualified as LBS
import Data.Either (fromRight)
import Data.Foldable (traverse_)
import Data.IORef (modifyIORef', newIORef, readIORef)
import Data.Int (Int64)
import Data.Kind (Type)
import Data.List (genericLength)
import Data.Map.Strict qualified as Map
import Data.Maybe (catMaybes, fromMaybe, isJust)
import Data.Ord (clamp)
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Text.Encoding (encodeUtf8)
import Data.Time (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime)
import Data.UUID.Types (UUID)
import Data.UUID.V4 qualified as UUID
import GHC.TypeLits (KnownNat, KnownSymbol, symbolVal)
import Network.HTTP.Types (hCacheControl, hContentType, status200)
import Network.Wai (responseStream)
import Network.Wai.Handler.Warp (Port, defaultSettings, runSettings, setPort)
import Servant
import Servant.Server.Generic (AsServerT)
import System.Timeout (timeout)
import UnliftIO.Exception (handleAny, tryAny)

import Arbiter.Servant.API
  ( ArbiterAPI
  , ArchiveAPI (..)
  , ConcurrencyAPI (..)
  , CronAPI (..)
  , DLQAPI (..)
  , HealthAPI (..)
  , JobsAPI (..)
  , MaintenanceAPI (..)
  , QueuesAPI (..)
  , RateLimitsAPI (..)
  , RegistryToAPI
  , SharedAPI
  , StatsAPI (..)
  , TableAPI (..)
  , WorkersAPI (..)
  )
import Arbiter.Servant.Types

-- | Configuration for the API server. @m@ is the backend monad every handler's
-- statements run in. 'createArbiterServer' fills the cache fields.
data ArbiterServerConfig m (registry :: JobPayloadRegistry) = ArbiterServerConfig
  { forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> forall a. m a -> IO a
serverRun :: forall a. m a -> IO a
  -- ^ Backend runner, such as @runSimpleDb env@ or @runHasqlDb env@.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema :: Text
  -- ^ The env's schema. 'createArbiterServer' reads it from the runner. Keep it equal
  -- to the env's schema.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Bool
enableSSE :: Bool
  -- ^ Enable the Server-Sent Events streaming endpoint. When 'False', the
  -- @\/events\/stream@ endpoint returns one \"disabled\" event and closes.
  -- A backend with no listener answers the same way.
  -- Default: 'True'.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
rateLimitPoliciesCache :: CacheCell RateLimitPoliciesResponse
  -- ^ Cache for the rate-limit policy list, used when 'statsCacheTtl' is positive.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
concurrencyPoliciesCache :: CacheCell ConcurrencyPoliciesResponse
  -- ^ Cache for the concurrency policy list, used when 'statsCacheTtl' is positive.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell AllStatsResponse
allQueueStatsCache :: CacheCell AllStatsResponse
  -- ^ Cache for the all-queues overview, used when 'statsCacheTtl' is positive.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell StatsResponse
queueStatsCache :: CacheCell StatsResponse
  -- ^ Per-queue stats cache, used when 'statsCacheTtl' is positive.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
statsCacheTtl :: NominalDiffTime
  -- ^ How old the queue stats, the overview and the policy lists can be. Zero
  -- reads the database on each request. Default: 'defaultStatsCacheTtl'.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell HealthResponse
healthCache :: CacheCell HealthResponse
  -- ^ Short-TTL cache for the readiness probe.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceInterval :: NominalDiffTime
  -- ^ Minimum gap between runs of one ordinary maintenance operation. Zero runs
  -- each ordinary operation on every call. Default: 'defaultMaintenanceInterval'.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceSparseInterval :: NominalDiffTime
  -- ^ Minimum gap between runs of the bucket prune and the concurrency
  -- reconcile-and-prune. Default: 'defaultMaintenanceSparseInterval'.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceBucketIdle :: NominalDiffTime
  -- ^ Idle age at which a pass prunes a full rate-limit bucket.
  -- Default: 'defaultMaintenanceBucketIdle'.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceTimeout :: NominalDiffTime
  -- ^ Abort any single maintenance statement that runs longer than this.
  -- Default: 'defaultMaintenanceTimeout'.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> LogConfig
serverLogConfig :: LogConfig
  -- ^ Where the server logs.
  -- Default: 'defaultLogConfig'.
  , forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> FailureGates
deadLetterGates :: FailureGates
  -- ^ Per-queue failure gates for dead-letter reports.
  }

-- | Run a statement on the backend.
runDb :: (MonadIO n) => ArbiterServerConfig m registry -> m a -> n a
runDb :: forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config = IO a -> n a
forall a. IO a -> n a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO a -> n a) -> (m a -> IO a) -> m a -> n a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ArbiterServerConfig m registry -> forall a. m a -> IO a
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> forall a. m a -> IO a
serverRun ArbiterServerConfig m registry
config

-- | 404 for a job id that matches no row.
jobNotFound :: ServerError
jobNotFound :: ServerError
jobNotFound = ServerError
err404 {errBody = "Job not found"}

-- | 404 for a DLQ id that matches no row.
dlqJobNotFound :: ServerError
dlqJobNotFound :: ServerError
dlqJobNotFound = ServerError
err404 {errBody = "DLQ job not found"}

-- | 404 for an archive id that matches no row.
archivedJobNotFound :: ServerError
archivedJobNotFound :: ServerError
archivedJobNotFound = ServerError
err404 {errBody = "Archived job not found"}

-- | 404 for a rate-limit prefix that matches no policy.
rateLimitPolicyNotFound :: ServerError
rateLimitPolicyNotFound :: ServerError
rateLimitPolicyNotFound = ServerError
err404 {errBody = "Rate-limit policy not found"}

-- | 404 for a cron name that matches no schedule.
cronScheduleNotFound :: ServerError
cronScheduleNotFound :: ServerError
cronScheduleNotFound = ServerError
err404 {errBody = "Cron schedule not found"}

-- | 404 for a worker id that matches no registry row.
workerNotFound :: ServerError
workerNotFound :: ServerError
workerNotFound = ServerError
err404 {errBody = "Worker not found"}

-- | 404 for a concurrency prefix that matches no policy.
concurrencyPolicyNotFound :: ServerError
concurrencyPolicyNotFound :: ServerError
concurrencyPolicyNotFound = ServerError
err404 {errBody = "Concurrency policy not found"}

-- | Answer a handler that decided its own error.
noContentOr :: Either ServerError () -> Handler NoContent
noContentOr :: Either ServerError () -> Handler NoContent
noContentOr = (ServerError -> Handler NoContent)
-> (() -> Handler NoContent)
-> Either ServerError ()
-> Handler NoContent
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either ServerError -> Handler NoContent
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError (Handler NoContent -> () -> Handler NoContent
forall a b. a -> b -> a
const (NoContent -> Handler NoContent
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure NoContent
NoContent))

-- | Run a mutation. When it touches no row, answer what @probe@ finds.
mutateOr
  :: (Monad m)
  => ArbiterServerConfig m registry
  -> (Text -> m Int64)
  -> (Text -> m (Either ServerError ()))
  -> Handler NoContent
mutateOr :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> (Text -> m Int64)
-> (Text -> m (Either ServerError ()))
-> Handler NoContent
mutateOr ArbiterServerConfig m registry
config Text -> m Int64
mutate Text -> m (Either ServerError ())
probe =
  Either ServerError () -> Handler NoContent
noContentOr (Either ServerError () -> Handler NoContent)
-> Handler (Either ServerError ()) -> Handler NoContent
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< ArbiterServerConfig m registry
-> m (Either ServerError ()) -> Handler (Either ServerError ())
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (Text -> m Int64
mutate Text
schemaName m Int64
-> (Int64 -> m (Either ServerError ()))
-> m (Either ServerError ())
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Int64 -> m (Either ServerError ())
diagnose)
  where
    schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
    diagnose :: Int64 -> m (Either ServerError ())
diagnose Int64
rowsAffected
      | Int64
rowsAffected Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
> Int64
0 = Either ServerError () -> m (Either ServerError ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either ServerError ()
forall a b. b -> Either a b
Right ())
      | Bool
otherwise = Text -> m (Either ServerError ())
probe Text
schemaName

-- | Run a mutation. When it touches no row, answer @missing@.
mutateOr404
  :: (Monad m)
  => ArbiterServerConfig m registry
  -> ServerError
  -> (Text -> m Int64)
  -> Handler NoContent
mutateOr404 :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
mutateOr404 ArbiterServerConfig m registry
config ServerError
missing Text -> m Int64
mutate =
  ArbiterServerConfig m registry
-> (Text -> m Int64)
-> (Text -> m (Either ServerError ()))
-> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> (Text -> m Int64)
-> (Text -> m (Either ServerError ()))
-> Handler NoContent
mutateOr ArbiterServerConfig m registry
config Text -> m Int64
mutate (m (Either ServerError ()) -> Text -> m (Either ServerError ())
forall a b. a -> b -> a
const (Either ServerError () -> m (Either ServerError ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerError -> Either ServerError ()
forall a b. a -> Either a b
Left ServerError
missing)))

-- | Run a job mutation. When it touches no row, re-read the job and answer 404, or
-- the 409 that @refuse@ derives from the job and its status.
mutateJob
  :: forall (payload :: Type) registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> (Text -> m Int64)
  -> (Job.JobRead (Job.Stored payload) -> JobStatus -> LBS.ByteString)
  -> Handler NoContent
mutateJob :: forall payload (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored payload) -> JobStatus -> ByteString)
-> Handler NoContent
mutateJob Text
tableName ArbiterServerConfig m registry
config Int64
jobId Text -> m Int64
mutate JobRead (Stored payload) -> JobStatus -> ByteString
refuse =
  ArbiterServerConfig m registry
-> (Text -> m Int64)
-> (Text -> m (Either ServerError ()))
-> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> (Text -> m Int64)
-> (Text -> m (Either ServerError ()))
-> Handler NoContent
mutateOr ArbiterServerConfig m registry
config Text -> m Int64
mutate ((Text -> m (Either ServerError ())) -> Handler NoContent)
-> (Text -> m (Either ServerError ())) -> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
    Either ServerError ()
-> ((JobRead (Stored payload), JobStatus) -> Either ServerError ())
-> Maybe (JobRead (Stored payload), JobStatus)
-> Either ServerError ()
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (ServerError -> Either ServerError ()
forall a b. a -> Either a b
Left ServerError
jobNotFound) (\(JobRead (Stored payload)
job, JobStatus
status) -> ServerError -> Either ServerError ()
forall a b. a -> Either a b
Left ServerError
err409 {errBody = refuse job status})
      (Maybe (JobRead (Stored payload), JobStatus)
 -> Either ServerError ())
-> m (Maybe (JobRead (Stored payload), JobStatus))
-> m (Either ServerError ())
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Int64 -> m (Maybe (JobRead (Stored payload), JobStatus))
Ops.getJobByIdWithStatus @_ @payload Text
schemaName Text
tableName Int64
jobId

-- | Run a mutation over a parent's subtree. When it changes no row, it succeeds, unless the parent is gone.
mutateChildren
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> (Text -> Text -> Int64 -> m Int64)
  -> Handler NoContent
mutateChildren :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> Text -> Int64 -> m Int64)
-> Handler NoContent
mutateChildren Text
tableName ArbiterServerConfig m registry
config Int64
jobId Text -> Text -> Int64 -> m Int64
mutate =
  ArbiterServerConfig m registry
-> (Text -> m Int64)
-> (Text -> m (Either ServerError ()))
-> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> (Text -> m Int64)
-> (Text -> m (Either ServerError ()))
-> Handler NoContent
mutateOr ArbiterServerConfig m registry
config (\Text
schemaName -> Text -> Text -> Int64 -> m Int64
mutate Text
schemaName Text
tableName Int64
jobId) ((Text -> m (Either ServerError ())) -> Handler NoContent)
-> (Text -> m (Either ServerError ())) -> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
    (\Bool
exists -> if Bool
exists then () -> Either ServerError ()
forall a b. b -> Either a b
Right () else ServerError -> Either ServerError ()
forall a b. a -> Either a b
Left ServerError
jobNotFound)
      (Bool -> Either ServerError ())
-> m Bool -> m (Either ServerError ())
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Text -> Int64 -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Bool
Ops.jobExists Text
schemaName Text
tableName Int64
jobId

-- | Create an 'ArbiterServerConfig' over a backend runner. SSE needs the
-- @enableEventStreaming@ migration option and a backend with a listener.
createArbiterServer
  :: forall m registry
   . (HasRegistry m registry)
  => (forall a. m a -> IO a)
  -> IO (ArbiterServerConfig m registry)
createArbiterServer :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
HasRegistry m registry =>
(forall a. m a -> IO a) -> IO (ArbiterServerConfig m registry)
createArbiterServer forall a. m a -> IO a
run = do
  schemaName <- m Text -> IO Text
forall a. m a -> IO a
run m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  rlCache <- newCacheCell
  ccCache <- newCacheCell
  statsCache <- newCacheCell
  perQueueCache <- newCacheCell
  healthCell <- newCacheCell
  gates <- newFailureGates
  pure
    ArbiterServerConfig
      { serverRun = run
      , serverSchema = schemaName
      , enableSSE = True
      , rateLimitPoliciesCache = rlCache
      , concurrencyPoliciesCache = ccCache
      , allQueueStatsCache = statsCache
      , queueStatsCache = perQueueCache
      , statsCacheTtl = defaultStatsCacheTtl
      , healthCache = healthCell
      , maintenanceInterval = defaultMaintenanceInterval
      , maintenanceSparseInterval = defaultMaintenanceSparseInterval
      , maintenanceBucketIdle = defaultMaintenanceBucketIdle
      , maintenanceTimeout = defaultMaintenanceTimeout
      , serverLogConfig = defaultLogConfig
      , deadLetterGates = gates
      }

-- | Jobs API handlers for a specific table.
jobsServer
  :: forall registry payload result m
   . (EncodeJobResult result, HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> JobsAPI payload result (AsServerT Handler)
jobsServer :: forall (registry :: JobPayloadRegistry) payload result
       (m :: * -> *).
(EncodeJobResult result, HasRegistry m registry,
 JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> JobsAPI payload result (AsServerT Handler)
jobsServer Text
table ArbiterServerConfig m registry
config =
  JobsAPI
    { listJobs :: AsServerT Handler
:- (Summary "List jobs"
    :> (Description
          "Filters combine. The payload filter searches the payload text."
        :> (QueryParam "limit" PageLimit
            :> (QueryParam "offset" Int
                :> (QueryParam "group_key" Text
                    :> (QueryParam "parent_id" Int64
                        :> (QueryParam "job_id" Int64
                            :> (QueryFlag "roots_only"
                                :> (QueryParam "status" JobStatus
                                    :> (QueryParam "claimed_by" UUID
                                        :> (QueryParam "kind" Text
                                            :> (QueryParam "payload" Text
                                                :> (QueryParam "rate_limit_prefix" Text
                                                    :> (QueryParam "concurrency_prefix" Text
                                                        :> (QueryParam "sort_by" JobSortColumn
                                                            :> (QueryParam "sort_dir" SortDir
                                                                :> Get
                                                                     '[JSON]
                                                                     (JobsResponse
                                                                        payload)))))))))))))))))
listJobs = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Text
-> Maybe Int64
-> Maybe Int64
-> Bool
-> Maybe JobStatus
-> Maybe UUID
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe JobSortColumn
-> Maybe SortDir
-> Handler (JobsResponse payload)
listJobsHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , insertJob :: AsServerT Handler
:- (Summary "Enqueue a job"
    :> (Description
          "A duplicate of an ignore dedup key returns the job that exists. A duplicate of a replace dedup key replaces the job in the queue."
        :> (Throws
              409
              "A replace dedup key matched a job that is in flight, flagged for cancel, or has children."
            :> (Throws
                  409
                  "An ignore dedup key matched a job that was deleted before the server could read it."
                :> (ReqBody '[JSON] (ApiJobWrite payload)
                    :> Post '[JSON] (JobResponse (JobRead payload)))))))
insertJob = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> ApiJobWrite payload
-> Handler (JobResponse (JobRead payload))
insertJobHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , insertJobsBatch :: AsServerT Handler
:- (Summary "Enqueue many jobs"
    :> (Description
          "Returns the jobs inserted or replaced. The response omits a job that an ignore dedup key skips, and a job that a replace dedup key cannot replace."
        :> ("batch"
            :> (ReqBody '[JSON] (BatchInsertRequest payload)
                :> Post '[JSON] (BatchInsertResponse payload)))))
insertJobsBatch = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> BatchInsertRequest payload
-> Handler (BatchInsertResponse payload)
insertJobsBatchHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , getJob :: AsServerT Handler
:- (Summary "Show a job"
    :> (Description "The response includes the derived status."
        :> (Capture "id" Int64
            :> Get '[JSON] (JobResponse (ApiJobWithStatus (Stored payload))))))
getJob = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> Handler (JobResponse (ApiJobWithStatus (Stored payload)))
getJobHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , cancelJob :: AsServerT Handler
:- (Summary "Cancel a job and its descendants"
    :> (Description
          "Deletes the job and every descendant, also the jobs in flight. The handler of a job in flight continues, and its ack finds no job. Use force-cancel to stop the handler."
        :> (Capture "id" Int64 :> DeleteNoContent)))
cancelJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
cancelJobHandler @registry Text
table ArbiterServerConfig m registry
config
    , forceCancelJob :: AsServerT Handler
:- (Summary "Cancel a job and interrupt its handlers"
    :> (Description
          "Deletes the jobs in the tree that are not in flight. A job in flight receives a cancel flag and cannot be claimed again. A worker pool stops its handler. An HTTP claimant with a live lease receives 409 on its next ack, nack or extend, until the reaper deletes the job. Then it receives 404. A claimant whose lease expired before the force-cancel receives 404."
        :> (Capture "id" Int64 :> ("force-cancel" :> PostNoContent))))
forceCancelJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
forceCancelJobHandler @registry Text
table ArbiterServerConfig m registry
config
    , promoteJob :: AsServerT Handler
:- (Summary "Make a job visible now"
    :> (Throws
          409
          "The job is already visible, suspended, in flight, or flagged for cancel."
        :> (Capture "id" Int64 :> ("promote" :> PostNoContent))))
promoteJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
promoteJobHandler @registry Text
table ArbiterServerConfig m registry
config
    , rescheduleJob :: AsServerT Handler
:- (Summary "Set when a job becomes visible"
    :> (Throws
          409
          "The job is in flight, suspended, flagged for cancel, or out of attempts, or another operation changed it."
        :> (Capture "id" Int64
            :> ("reschedule"
                :> (ReqBody '[JSON] RescheduleRequest :> PostNoContent)))))
rescheduleJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> RescheduleRequest
-> Handler NoContent
rescheduleJobHandler @registry Text
table ArbiterServerConfig m registry
config
    , moveToDLQ :: AsServerT Handler
:- (Summary "Move a job to the DLQ"
    :> (Description
          "A rollup takes its descendants with it. A job in flight is moved too. Its handler continues, and its ack finds no job."
        :> (Throws
              409
              "Another operation changed or deleted the job between the read and the move."
            :> (Capture "id" Int64 :> ("move-to-dlq" :> PostNoContent)))))
moveToDLQ = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
moveToDLQHandler @registry Text
table ArbiterServerConfig m registry
config
    , pauseChildren :: AsServerT Handler
:- (Summary "Suspend the visible descendants of a job"
    :> (Description
          "The operation skips descendants that are in flight, delayed or throttled. Succeeds when nothing is suspended. 404 when the job is gone."
        :> (Capture "id" Int64 :> ("pause-children" :> PostNoContent))))
pauseChildren = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
pauseChildrenHandler @registry Text
table ArbiterServerConfig m registry
config
    , resumeChildren :: AsServerT Handler
:- (Summary "Resume the suspended descendants of a job"
    :> (Description
          "A finalizer with children in the queue stays suspended. Succeeds when nothing is resumed. 404 when the job is gone."
        :> (Capture "id" Int64 :> ("resume-children" :> PostNoContent))))
resumeChildren = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
resumeChildrenHandler @registry Text
table ArbiterServerConfig m registry
config
    , suspendJob :: AsServerT Handler
:- (Summary "Suspend a job"
    :> (Description "A suspended job cannot be claimed."
        :> (Throws 409 "The job is already suspended, or it is in flight."
            :> (Capture "id" Int64 :> ("suspend" :> PostNoContent)))))
suspendJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
suspendJobHandler @registry Text
table ArbiterServerConfig m registry
config
    , resumeJob :: AsServerT Handler
:- (Summary "Resume a suspended job"
    :> (Throws
          409
          "The job is not suspended, it is a finalizer with children in the queue, or another operation changed it."
        :> (Capture "id" Int64 :> ("resume" :> PostNoContent))))
resumeJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
resumeJobHandler @registry Text
table ArbiterServerConfig m registry
config
    , ackClaimedJob :: AsServerT Handler
:- (Summary "Complete a held job"
    :> (Description
          "Send the claimSeq and claimedBy from the claim. The server stores the optional result for the parent rollup or the archive."
        :> (Throws 409 LeaseRefusal
            :> (Capture "id" Int64
                :> ("ack"
                    :> (ReqBody '[JSON] (AckRequest result) :> PostNoContent))))))
ackClaimedJob = forall (registry :: JobPayloadRegistry) result (m :: * -> *).
(EncodeJobResult result, HasRegistry m registry) =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> AckRequest result
-> Handler NoContent
ackClaimedJobHandler @registry @result Text
table ArbiterServerConfig m registry
config
    , nackClaimedJob :: AsServerT Handler
:- (Summary "Return a held job"
    :> (Description
          "Restores the attempt that the claim used. The job becomes claimable when its lease expires."
        :> (Throws 409 LeaseRefusal
            :> (Capture "id" Int64
                :> ("nack" :> (ReqBody '[JSON] JobLease :> PostNoContent))))))
nackClaimedJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> JobLease
-> Handler NoContent
nackClaimedJobHandler @registry Text
table ArbiterServerConfig m registry
config
    , extendClaimedJob :: AsServerT Handler
:- (Summary "Extend a held lease"
    :> (Description "Sets the lease to expire leaseSeconds from now."
        :> (Throws 409 LeaseRefusal
            :> (Capture "id" Int64
                :> ("extend"
                    :> (ReqBody '[JSON] ExtendRequest :> PostNoContent))))))
extendClaimedJob = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> ExtendRequest
-> Handler NoContent
extendClaimedJobHandler @registry Text
table ArbiterServerConfig m registry
config
    }

-- | List jobs with pagination and composable filters.
listJobsHandler
  :: forall registry (payload :: Type) m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Maybe PageLimit
  -> Maybe Int
  -> Maybe Text
  -> Maybe Int64
  -> Maybe Int64
  -> Bool
  -> Maybe JobStatus
  -> Maybe UUID
  -> Maybe Text
  -> Maybe Text
  -> Maybe Text
  -> Maybe Text
  -> Maybe JobSortColumn
  -> Maybe SortDir
  -> Handler (JobsResponse payload)
listJobsHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Text
-> Maybe Int64
-> Maybe Int64
-> Bool
-> Maybe JobStatus
-> Maybe UUID
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe JobSortColumn
-> Maybe SortDir
-> Handler (JobsResponse payload)
listJobsHandler Text
tableName ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset Maybe Text
mGroupKey Maybe Int64
mParentId Maybe Int64
mJobId Bool
rootsOnly Maybe JobStatus
mStatus Maybe UUID
mClaimedBy Maybe Text
mKind Maybe Text
mPayload Maybe Text
mRatePrefix Maybe Text
mConcPrefix Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir = IO (JobsResponse payload) -> Handler (JobsResponse payload)
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (JobsResponse payload) -> Handler (JobsResponse payload))
-> IO (JobsResponse payload) -> Handler (JobsResponse payload)
forall a b. (a -> b) -> a -> b
$ do
  let (Int
limit, Int
offset) = Maybe PageLimit -> Maybe Int -> (Int, Int)
forall (def :: Nat).
KnownNat def =>
Maybe (PageSize def) -> Maybe Int -> (Int, Int)
validatePagination Maybe PageLimit
mLimit Maybe Int
mOffset
      schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
      filters :: [JobFilter]
filters =
        [Maybe JobFilter] -> [JobFilter]
forall a. [Maybe a] -> [a]
catMaybes ([Maybe JobFilter] -> [JobFilter])
-> [Maybe JobFilter] -> [JobFilter]
forall a b. (a -> b) -> a -> b
$
          Maybe Int64
-> Maybe Text -> Maybe Text -> Maybe Text -> [Maybe JobFilter]
listingFilters Maybe Int64
mParentId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload
            [Maybe JobFilter] -> [Maybe JobFilter] -> [Maybe JobFilter]
forall a. Semigroup a => a -> a -> a
<> [ Int64 -> JobFilter
FilterId (Int64 -> JobFilter) -> Maybe Int64 -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Int64
mJobId
               , JobFilter
FilterRootsOnly JobFilter -> Maybe () -> Maybe JobFilter
forall a b. a -> Maybe b -> Maybe a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ Bool -> Maybe ()
forall (f :: * -> *). Alternative f => Bool -> f ()
guard Bool
rootsOnly
               , JobStatus -> JobFilter
FilterStatus (JobStatus -> JobFilter) -> Maybe JobStatus -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe JobStatus
mStatus
               , UUID -> JobFilter
FilterClaimedBy (UUID -> JobFilter) -> Maybe UUID -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe UUID
mClaimedBy
               , Text -> JobFilter
FilterRateLimitPrefix (Text -> JobFilter) -> Maybe Text -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text
mRatePrefix
               , Text -> JobFilter
FilterConcurrencyPrefix (Text -> JobFilter) -> Maybe Text -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text
mConcPrefix
               ]

  (jobs, total, combined, dlqCounts) <- ArbiterServerConfig m registry
-> m ([(JobRead (Stored payload), JobStatus)], Int64,
      Map Int64 (Int64, Int64), Map Int64 Int64)
-> IO
     ([(JobRead (Stored payload), JobStatus)], Int64,
      Map Int64 (Int64, Int64), Map Int64 Int64)
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m ([(JobRead (Stored payload), JobStatus)], Int64,
    Map Int64 (Int64, Int64), Map Int64 Int64)
 -> IO
      ([(JobRead (Stored payload), JobStatus)], Int64,
       Map Int64 (Int64, Int64), Map Int64 Int64))
-> m ([(JobRead (Stored payload), JobStatus)], Int64,
      Map Int64 (Int64, Int64), Map Int64 Int64)
-> IO
     ([(JobRead (Stored payload), JobStatus)], Int64,
      Map Int64 (Int64, Int64), Map Int64 Int64)
forall a b. (a -> b) -> a -> b
$ m ([(JobRead (Stored payload), JobStatus)], Int64,
   Map Int64 (Int64, Int64), Map Int64 Int64)
-> m ([(JobRead (Stored payload), JobStatus)], Int64,
      Map Int64 (Int64, Int64), Map Int64 Int64)
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m ([(JobRead (Stored payload), JobStatus)], Int64,
    Map Int64 (Int64, Int64), Map Int64 Int64)
 -> m ([(JobRead (Stored payload), JobStatus)], Int64,
       Map Int64 (Int64, Int64), Map Int64 Int64))
-> m ([(JobRead (Stored payload), JobStatus)], Int64,
      Map Int64 (Int64, Int64), Map Int64 Int64)
-> m ([(JobRead (Stored payload), JobStatus)], Int64,
      Map Int64 (Int64, Int64), Map Int64 Int64)
forall a b. (a -> b) -> a -> b
$ do
    page <- Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead (Stored payload), JobStatus)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe JobSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [(JobRead (Stored payload), JobStatus)]
Ops.listJobsWithStatus Text
schemaName Text
tableName [JobFilter]
filters Maybe JobSortColumn
mSortBy Maybe SortDir
mSortDir Int
limit Int
offset
    matching <- Ops.countJobsFiltered schemaName tableName filters
    -- Every parent is a rollup finalizer. A page without one skips the count queries.
    let jobIds = ((JobRead (Stored payload), JobStatus) -> Int64)
-> [(JobRead (Stored payload), JobStatus)] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead (Stored payload) -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
Job.primaryKey (JobRead (Stored payload) -> Int64)
-> ((JobRead (Stored payload), JobStatus)
    -> JobRead (Stored payload))
-> (JobRead (Stored payload), JobStatus)
-> Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead (Stored payload), JobStatus) -> JobRead (Stored payload)
forall a b. (a, b) -> a
fst) [(JobRead (Stored payload), JobStatus)]
page
        hasParents = ((JobRead (Stored payload), JobStatus) -> Bool)
-> [(JobRead (Stored payload), JobStatus)] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (JobRead (Stored payload) -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
isRollup (JobRead (Stored payload) -> Bool)
-> ((JobRead (Stored payload), JobStatus)
    -> JobRead (Stored payload))
-> (JobRead (Stored payload), JobStatus)
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (JobRead (Stored payload), JobStatus) -> JobRead (Stored payload)
forall a b. (a, b) -> a
fst) [(JobRead (Stored payload), JobStatus)]
page
    if hasParents
      then do
        children <- Ops.countChildrenBatch schemaName tableName jobIds
        dlqChildren <- Ops.countDLQChildrenBatch schemaName tableName jobIds
        pure (page, matching, children, dlqChildren)
      else pure (page, matching, Map.empty, Map.empty)

  let childCounts = ((Int64, Int64) -> Int64)
-> Map Int64 (Int64, Int64) -> Map Int64 Int64
forall a b. (a -> b) -> Map Int64 a -> Map Int64 b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (Int64, Int64) -> Int64
forall a b. (a, b) -> a
fst Map Int64 (Int64, Int64)
combined
      pausedParents = Map Int64 (Int64, Int64) -> [Int64]
forall k a. Map k a -> [k]
Map.keys (Map Int64 (Int64, Int64) -> [Int64])
-> Map Int64 (Int64, Int64) -> [Int64]
forall a b. (a -> b) -> a -> b
$ ((Int64, Int64) -> Bool)
-> Map Int64 (Int64, Int64) -> Map Int64 (Int64, Int64)
forall a k. (a -> Bool) -> Map k a -> Map k a
Map.filter (\(Int64
childTotal, Int64
childPaused) -> Int64
childPaused Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
== Int64
childTotal) Map Int64 (Int64, Int64)
combined
      apiJobs = ((JobRead (Stored payload), JobStatus)
 -> ApiJobWithStatus (Stored payload))
-> [(JobRead (Stored payload), JobStatus)]
-> [ApiJobWithStatus (Stored payload)]
forall a b. (a -> b) -> [a] -> [b]
map (\(JobRead (Stored payload)
found, JobStatus
jobStatus) -> ApiJobWithStatus {ajwsJob :: JobRead (Stored payload)
ajwsJob = JobRead (Stored payload)
found, ajwsStatus :: JobStatus
ajwsStatus = JobStatus
jobStatus}) [(JobRead (Stored payload), JobStatus)]
jobs
  pure $
    JobsResponse
      { jobsPage = toPage (limit, offset) total apiJobs
      , childCounts = childCounts
      , pausedParents = pausedParents
      , dlqChildCounts = dlqCounts
      }

-- | Insert a new job into the queue.
insertJobHandler
  :: forall registry payload m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> ApiJobWrite payload
  -> Handler (JobResponse (Job.JobRead payload))
insertJobHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> ApiJobWrite payload
-> Handler (JobResponse (JobRead payload))
insertJobHandler Text
tableName ArbiterServerConfig m registry
config (ApiJobWrite JobWrite payload
jobWrite) = do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
  found <- ArbiterServerConfig m registry
-> m (Either ServerError (JobRead payload))
-> Handler (Either ServerError (JobRead payload))
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m (Either ServerError (JobRead payload))
 -> Handler (Either ServerError (JobRead payload)))
-> m (Either ServerError (JobRead payload))
-> Handler (Either ServerError (JobRead payload))
forall a b. (a -> b) -> a -> b
$ Text
-> [JobWrite payload]
-> m (Either ServerError (JobRead payload))
-> m (Either ServerError (JobRead payload))
forall payload (m :: * -> *) a.
(HasKind payload, MonadUnliftIO m) =>
Text -> [JobWrite payload] -> m a -> m a
withPublishSpan Text
tableName [JobWrite payload
jobWrite] (m (Either ServerError (JobRead payload))
 -> m (Either ServerError (JobRead payload)))
-> m (Either ServerError (JobRead payload))
-> m (Either ServerError (JobRead payload))
forall a b. (a -> b) -> a -> b
$ do
    inserted <- 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
jobWrite
    case (inserted, Job.dedupKey jobWrite) of
      (Just JobRead payload
fresh, Maybe DedupKey
_) -> Either ServerError (JobRead payload)
-> m (Either ServerError (JobRead payload))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (JobRead payload -> Either ServerError (JobRead payload)
forall a b. b -> Either a b
Right JobRead payload
fresh)
      (Maybe (JobRead payload)
Nothing, Just (IgnoreDuplicate Text
duplicateKey)) ->
        Either ServerError (JobRead payload)
-> (JobRead payload -> Either ServerError (JobRead payload))
-> Maybe (JobRead payload)
-> Either ServerError (JobRead payload)
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (ServerError -> Either ServerError (JobRead payload)
forall a b. a -> Either a b
Left ServerError
vanished) JobRead payload -> Either ServerError (JobRead payload)
forall a b. b -> Either a b
Right
          (Maybe (JobRead payload) -> Either ServerError (JobRead payload))
-> m (Maybe (JobRead payload))
-> m (Either ServerError (JobRead payload))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (Text -> Text -> Text -> m (Maybe (JobRead (Stored payload)))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Text -> m (Maybe (JobRead (Stored payload)))
Ops.getJobByDedupKey Text
schemaName Text
tableName Text
duplicateKey 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)
      (Maybe (JobRead payload), Maybe DedupKey)
_ -> Either ServerError (JobRead payload)
-> m (Either ServerError (JobRead payload))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerError -> Either ServerError (JobRead payload)
forall a b. a -> Either a b
Left ServerError
blocked)
  either throwError (\JobRead payload
fresh -> JobResponse (JobRead payload)
-> Handler (JobResponse (JobRead payload))
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure JobResponse {job :: JobRead payload
job = JobRead payload
fresh}) found
  where
    vanished :: ServerError
vanished = ServerError
err409 {errBody = "Duplicate job was deleted before it could be read"}
    blocked :: ServerError
blocked = ServerError
err409 {errBody = "Replace blocked: existing job is actively claimed, force-cancel flagged, or has children"}

-- | Insert multiple jobs in a single batch operation.
insertJobsBatchHandler
  :: forall registry payload m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> BatchInsertRequest payload
  -> Handler (BatchInsertResponse payload)
insertJobsBatchHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> BatchInsertRequest payload
-> Handler (BatchInsertResponse payload)
insertJobsBatchHandler Text
tableName ArbiterServerConfig m registry
config (BatchInsertRequest [ApiJobWrite payload]
apiWrites) = do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
      writes :: [JobWrite payload]
writes = (ApiJobWrite payload -> JobWrite payload)
-> [ApiJobWrite payload] -> [JobWrite payload]
forall a b. (a -> b) -> [a] -> [b]
map ApiJobWrite payload -> JobWrite payload
forall payload. ApiJobWrite payload -> JobWrite payload
unApiJobWrite [ApiJobWrite payload]
apiWrites

  inserted <-
    ArbiterServerConfig m registry
-> m [JobRead payload] -> Handler [JobRead payload]
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config
      (m [JobRead payload] -> Handler [JobRead payload])
-> m [JobRead payload] -> Handler [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Text
-> [JobWrite payload] -> m [JobRead payload] -> m [JobRead payload]
forall payload (m :: * -> *) a.
(HasKind payload, MonadUnliftIO m) =>
Text -> [JobWrite payload] -> m a -> m a
withPublishSpan Text
tableName [JobWrite payload]
writes
      (m [JobRead payload] -> m [JobRead payload])
-> m [JobRead payload] -> m [JobRead payload]
forall a b. (a -> b) -> a -> b
$ 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]
writes
  pure $ BatchInsertResponse {inserted = inserted, insertedCount = length inserted}

-- | Fetch a job by id.
getJobHandler
  :: forall registry (payload :: Type) m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler (JobResponse (ApiJobWithStatus (Job.Stored payload)))
getJobHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> Handler (JobResponse (ApiJobWithStatus (Stored payload)))
getJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  (\(JobRead (Stored payload)
found, JobStatus
jobStatus) -> JobResponse {job :: ApiJobWithStatus (Stored payload)
job = ApiJobWithStatus {ajwsJob :: JobRead (Stored payload)
ajwsJob = JobRead (Stored payload)
found, ajwsStatus :: JobStatus
ajwsStatus = JobStatus
jobStatus}})
    ((JobRead (Stored payload), JobStatus)
 -> JobResponse (ApiJobWithStatus (Stored payload)))
-> Handler (JobRead (Stored payload), JobStatus)
-> Handler (JobResponse (ApiJobWithStatus (Stored payload)))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry
-> m (Maybe (JobRead (Stored payload), JobStatus))
-> ServerError
-> Handler (JobRead (Stored payload), JobStatus)
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> m (Maybe a) -> ServerError -> Handler a
orNotFound ArbiterServerConfig m registry
config (Text
-> Text -> Int64 -> m (Maybe (JobRead (Stored payload), JobStatus))
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text -> Int64 -> m (Maybe (JobRead (Stored payload), JobStatus))
Ops.getJobByIdWithStatus (ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config) Text
tableName Int64
jobId) ServerError
jobNotFound

-- | Cancel a job (delete it from the queue).
cancelJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
cancelJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
cancelJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
mutateOr404 ArbiterServerConfig m registry
config ServerError
jobNotFound ((Text -> m Int64) -> Handler NoContent)
-> (Text -> m Int64) -> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
    [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

-- | Cascade-cancel a job and async-cancel any in-flight handlers via NOTIFY.
forceCancelJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
forceCancelJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
forceCancelJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
mutateOr404 ArbiterServerConfig m registry
config ServerError
jobNotFound ((Text -> m Int64) -> Handler NoContent)
-> (Text -> m Int64) -> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.forceCancelJob Text
schemaName Text
tableName Int64
jobId

-- | Promote a job (make it immediately visible).
promoteJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
promoteJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
promoteJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored (ZonkAny 0)) -> JobStatus -> ByteString)
-> Handler NoContent
forall payload (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored payload) -> JobStatus -> ByteString)
-> Handler NoContent
mutateJob Text
tableName ArbiterServerConfig m registry
config Int64
jobId (\Text
schemaName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.promoteJob Text
schemaName Text
tableName Int64
jobId) JobRead (Stored (ZonkAny 0)) -> JobStatus -> ByteString
forall {p}. JobRead p -> JobStatus -> ByteString
refuse
  where
    refuse :: JobRead p -> JobStatus -> ByteString
refuse JobRead p
job JobStatus
status = ByteString -> Maybe ByteString -> ByteString
forall a. a -> Maybe a -> a
fromMaybe ByteString
"Job is already visible" (JobRead p -> JobStatus -> Maybe ByteString
forall p. JobRead p -> JobStatus -> Maybe ByteString
heldReason JobRead p
job JobStatus
status)

-- | Why a job cannot be moved in time: it is suspended, in flight, or flagged for cancel.
heldReason :: Job.JobRead p -> JobStatus -> Maybe LBS.ByteString
heldReason :: forall p. JobRead p -> JobStatus -> Maybe ByteString
heldReason JobRead p
job JobStatus
status
  | JobRead p -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
Job.suspended JobRead p
job = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"Job is suspended - use resume endpoint"
  | JobStatus
status JobStatus -> JobStatus -> Bool
forall a. Eq a => a -> a -> Bool
== JobStatus
Job.InFlight = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"Job is in flight - wait for its lease to lapse"
  | JobStatus
status JobStatus -> JobStatus -> Bool
forall a. Eq a => a -> a -> Bool
== JobStatus
Job.Cancelled = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"Job is cancelled - it waits for removal"
  | Bool
otherwise = Maybe ByteString
forall a. Maybe a
Nothing

-- | Set when a job next becomes visible. Refuses an in-flight, suspended,
-- cancel-flagged or exhausted job.
rescheduleJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> RescheduleRequest
  -> Handler NoContent
rescheduleJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> RescheduleRequest
-> Handler NoContent
rescheduleJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId RescheduleRequest
request =
  Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored (ZonkAny 1)) -> JobStatus -> ByteString)
-> Handler NoContent
forall payload (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored payload) -> JobStatus -> ByteString)
-> Handler NoContent
mutateJob Text
tableName ArbiterServerConfig m registry
config Int64
jobId (\Text
schemaName -> Text -> Text -> Int64 -> UTCTime -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> UTCTime -> m Int64
Ops.rescheduleJob Text
schemaName Text
tableName Int64
jobId (RescheduleRequest -> UTCTime
runAt RescheduleRequest
request)) JobRead (Stored (ZonkAny 1)) -> JobStatus -> ByteString
forall {p}. JobRead p -> JobStatus -> ByteString
refuse
  where
    refuse :: JobRead p -> JobStatus -> ByteString
refuse JobRead p
job JobStatus
status
      | JobStatus
status JobStatus -> JobStatus -> Bool
forall a. Eq a => a -> a -> Bool
== JobStatus
Job.Exhausted = ByteString
"Job is exhausted - it waits for the dead letter queue"
      | Bool
otherwise = ByteString -> Maybe ByteString -> ByteString
forall a. a -> Maybe a -> a
fromMaybe ByteString
"Job could not be rescheduled (concurrent modification)" (JobRead p -> JobStatus -> Maybe ByteString
forall p. JobRead p -> JobStatus -> Maybe ByteString
heldReason JobRead p
job JobStatus
status)

-- | Move a job to the dead letter queue.
moveToDLQHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
moveToDLQHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
moveToDLQHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  Either ServerError () -> Handler NoContent
noContentOr (Either ServerError () -> Handler NoContent)
-> Handler (Either ServerError ()) -> Handler NoContent
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< ArbiterServerConfig m registry
-> m (Either ServerError ()) -> Handler (Either ServerError ())
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m (Either ServerError ()) -> m (Either ServerError ())
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction m (Either ServerError ())
moved)
  where
    schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
    moved :: m (Either ServerError ())
moved = Text -> Text -> Int64 -> m (Maybe (JobRead (Stored (ZonkAny 4))))
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 (ZonkAny 4))))
-> (Maybe (JobRead (Stored (ZonkAny 4)))
    -> m (Either ServerError ()))
-> m (Either ServerError ())
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= m (Either ServerError ())
-> (JobRead (Stored (ZonkAny 4)) -> m (Either ServerError ()))
-> Maybe (JobRead (Stored (ZonkAny 4)))
-> m (Either ServerError ())
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (Either ServerError () -> m (Either ServerError ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Either ServerError ()
forall {b}. Either ServerError b
notFound) JobRead (Stored (ZonkAny 4)) -> m (Either ServerError ())
move
    notFound :: Either ServerError b
notFound = ServerError -> Either ServerError b
forall a b. a -> Either a b
Left ServerError
jobNotFound
    move :: JobRead (Stored (ZonkAny 4)) -> m (Either ServerError ())
move JobRead (Stored (ZonkAny 4))
job = Int64 -> Either ServerError ()
forall {a}. (Ord a, Num a) => a -> Either ServerError ()
decide (Int64 -> Either ServerError ())
-> m Int64 -> m (Either ServerError ())
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TreeLocks
-> Text -> Text -> Text -> JobRead (Stored (ZonkAny 4)) -> 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
"Manually moved to DLQ via admin API" JobRead (Stored (ZonkAny 4))
job
    decide :: a -> Either ServerError ()
decide a
rowsAffected
      | a
rowsAffected a -> a -> Bool
forall a. Ord a => a -> a -> Bool
> a
0 = () -> Either ServerError ()
forall a b. b -> Either a b
Right ()
      | Bool
otherwise = ServerError -> Either ServerError ()
forall a b. a -> Either a b
Left ServerError
err409 {errBody = "Job was concurrently modified"}

-- | Pause all children of a parent job.
pauseChildrenHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
pauseChildrenHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
pauseChildrenHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> Text -> Int64 -> m Int64)
-> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> Text -> Int64 -> m Int64)
-> Handler NoContent
mutateChildren Text
tableName ArbiterServerConfig m registry
config Int64
jobId Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.pauseChildren

-- | Resume all suspended children of a parent job.
resumeChildrenHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
resumeChildrenHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
resumeChildrenHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> Text -> Int64 -> m Int64)
-> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> Text -> Int64 -> m Int64)
-> Handler NoContent
mutateChildren Text
tableName ArbiterServerConfig m registry
config Int64
jobId Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.resumeChildren

-- | Suspend a job (make it unclaimable).
suspendJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
suspendJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
suspendJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored (ZonkAny 2)) -> JobStatus -> ByteString)
-> Handler NoContent
forall payload (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored payload) -> JobStatus -> ByteString)
-> Handler NoContent
mutateJob Text
tableName ArbiterServerConfig m registry
config Int64
jobId (\Text
schemaName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.suspendJob Text
schemaName Text
tableName Int64
jobId) JobRead (Stored (ZonkAny 2)) -> JobStatus -> ByteString
forall {a} {payload} {q} {insertedAt} {adm} {p}.
IsString a =>
Job payload Int64 q insertedAt adm -> p -> a
refuse
  where
    refuse :: Job payload Int64 q insertedAt adm -> p -> a
refuse Job payload Int64 q insertedAt adm
job p
_
      | Job payload Int64 q insertedAt adm -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
Job.suspended Job payload Int64 q insertedAt adm
job = a
"Job is already suspended"
      | Bool
otherwise = a
"Job is in flight - cannot suspend"

-- | Resume a suspended job, making it claimable again. Refuses a finalizer with children
-- still running.
resumeJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
resumeJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
resumeJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId =
  Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored (ZonkAny 3)) -> JobStatus -> ByteString)
-> Handler NoContent
forall payload (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> (Text -> m Int64)
-> (JobRead (Stored payload) -> JobStatus -> ByteString)
-> Handler NoContent
mutateJob Text
tableName ArbiterServerConfig m registry
config Int64
jobId (\Text
schemaName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.resumeJob Text
schemaName Text
tableName Int64
jobId) JobRead (Stored (ZonkAny 3)) -> JobStatus -> ByteString
forall {a} {payload} {q} {insertedAt} {adm} {p}.
IsString a =>
Job payload Int64 q insertedAt adm -> p -> a
refuse
  where
    refuse :: Job p Int64 q t adm -> p -> a
refuse Job p Int64 q t adm
job p
_
      | Bool -> Bool
not (Job p Int64 q t adm -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
Job.suspended Job p Int64 q t adm
job) = a
"Job is not suspended"
      | Job p Int64 q t adm -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
isRollup Job p Int64 q t adm
job = a
"Cannot resume a rollup finalizer with active children"
      | Bool
otherwise = a
"Job could not be resumed (concurrent modification)"

-- | DLQ API handlers for a specific table.
dlqServer
  :: forall registry payload m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> DLQAPI payload (AsServerT Handler)
dlqServer :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> DLQAPI payload (AsServerT Handler)
dlqServer Text
table ArbiterServerConfig m registry
config =
  DLQAPI
    { listDLQ :: AsServerT Handler
:- (Summary "List dead-lettered jobs"
    :> (Description
          "The payload and error filters search the payload and the last error."
        :> (QueryParam "limit" PageLimit
            :> (QueryParam "offset" Int
                :> (QueryParam "parent_id" Int64
                    :> (QueryParam "job_id" Int64
                        :> (QueryParam "group_key" Text
                            :> (QueryParam "kind" Text
                                :> (QueryParam "payload" Text
                                    :> (QueryParam "error" Text
                                        :> (QueryParam "sort_by" DLQSortColumn
                                            :> (QueryParam "sort_dir" SortDir
                                                :> Get '[JSON] (DLQResponse payload)))))))))))))
listDLQ = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Int64
-> Maybe Int64
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Handler (DLQResponse payload)
listDLQHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , retryFromDLQ :: AsServerT Handler
:- (Summary "Retry a dead-lettered job"
    :> (Description "Moves the job back to the queue."
        :> (Throws
              409 "The parent of the job is gone from the queue and the DLQ."
            :> (Capture "id" Int64
                :> ("retry"
                    :> (ReqBody'
                          '[Description PayloadEditNote]
                          '[OptionalJSON]
                          (Maybe (PayloadEdit payload))
                        :> PostNoContent))))))
retryFromDLQ = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> Maybe (PayloadEdit payload)
-> Handler NoContent
retryFromDLQHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , deleteDLQ :: AsServerT Handler
:- (Summary "Delete a dead-lettered job"
    :> (Description
          "Resumes its parent when no child of the parent is left in the queue."
        :> (Capture "id" Int64 :> DeleteNoContent)))
deleteDLQ = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
deleteDLQHandler @registry Text
table ArbiterServerConfig m registry
config
    , deleteDLQBatch :: AsServerT Handler
:- (Summary "Delete dead-lettered jobs"
    :> (Description
          "Returns the number deleted. The server skips unknown ids."
        :> ("batch-delete"
            :> (ReqBody '[JSON] BatchDeleteRequest
                :> Post '[JSON] BatchDeleteResponse))))
deleteDLQBatch = ArbiterServerConfig m registry
-> (Text -> [Int64] -> m Int64)
-> BatchDeleteRequest
-> Handler BatchDeleteResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> (Text -> [Int64] -> m Int64)
-> BatchDeleteRequest
-> Handler BatchDeleteResponse
deleteBatchHandler ArbiterServerConfig m registry
config (Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
`Ops.deleteDLQJobsBatch` Text
table)
    , retryDLQBatch :: AsServerT Handler
:- (Summary "Retry dead-lettered jobs"
    :> (Description
          "Retries each job's tree in its own transaction. Returns the number of ids whose retry restored a tree. The server skips unknown ids and ids restored with an earlier id's tree."
        :> ("batch-retry"
            :> (ReqBody '[JSON] BatchRetryRequest
                :> Post '[JSON] BatchRetryResponse))))
retryDLQBatch = \(BatchRetryRequest [Int64]
dlqIds) ->
        ArbiterServerConfig m registry
-> (Text -> m Int64) -> Handler BatchRetryResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> (Text -> m Int64) -> Handler BatchRetryResponse
retriedResponse ArbiterServerConfig m registry
config (\Text
schemaName -> Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
Ops.retryFromDLQBatch Text
schemaName Text
table [Int64]
dlqIds)
    , retryDLQMatching :: AsServerT Handler
:- (Summary "Retry every matching dead-lettered job"
    :> (Description
          "Retries every job the filters match, oldest first, as batch-retry does. With no filters it retries the whole DLQ. The payload and error filters search the payload and the last error."
        :> ("retry-matching"
            :> (QueryParam "parent_id" Int64
                :> (QueryParam "job_id" Int64
                    :> (QueryParam "group_key" Text
                        :> (QueryParam "kind" Text
                            :> (QueryParam "payload" Text
                                :> (QueryParam "error" Text
                                    :> Post '[JSON] BatchRetryResponse)))))))))
retryDLQMatching = \Maybe Int64
mParentId Maybe Int64
mJobId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload Maybe Text
mError ->
        ArbiterServerConfig m registry
-> (Text -> m Int64) -> Handler BatchRetryResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> (Text -> m Int64) -> Handler BatchRetryResponse
retriedResponse ArbiterServerConfig m registry
config ((Text -> m Int64) -> Handler BatchRetryResponse)
-> (Text -> m Int64) -> Handler BatchRetryResponse
forall a b. (a -> b) -> a -> b
$ \Text
schemaName ->
          Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
Ops.retryDLQFiltered Text
schemaName Text
table (Maybe Int64
-> Maybe Int64
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> [JobFilter]
dlqFilters Maybe Int64
mParentId Maybe Int64
mJobId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload Maybe Text
mError)
    }

-- | List DLQ jobs with pagination and composable filters.
listDLQHandler
  :: forall registry (payload :: Type) m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Maybe PageLimit
  -> Maybe Int
  -> Maybe Int64
  -> Maybe Int64
  -> Maybe Text
  -> Maybe Text
  -> Maybe Text
  -> Maybe Text
  -> Maybe DLQSortColumn
  -> Maybe SortDir
  -> Handler (DLQResponse payload)
listDLQHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Int64
-> Maybe Int64
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Handler (DLQResponse payload)
listDLQHandler Text
tableName ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset Maybe Int64
mParentId Maybe Int64
mJobId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload Maybe Text
mError Maybe DLQSortColumn
mSortBy Maybe SortDir
mSortDir =
  ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> (Int -> Int -> m [DLQJob (Stored payload)])
-> m Int64
-> Handler (Page (DLQJob (Stored payload)))
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
MonadArbiter m =>
ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> (Int -> Int -> m [a])
-> m Int64
-> Handler (Page a)
readPage ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset Int -> Int -> m [DLQJob (Stored payload)]
page (Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
Ops.countDLQFiltered Text
schemaName Text
tableName [JobFilter]
filters)
  where
    schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
    page :: Int -> Int -> m [DLQJob (Stored payload)]
page = Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob (Stored payload)]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text
-> Text
-> [JobFilter]
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Int
-> Int
-> m [DLQJob (Stored payload)]
Ops.listDLQFilteredOrdered Text
schemaName Text
tableName [JobFilter]
filters Maybe DLQSortColumn
mSortBy Maybe SortDir
mSortDir
    filters :: [JobFilter]
filters = Maybe Int64
-> Maybe Int64
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> [JobFilter]
dlqFilters Maybe Int64
mParentId Maybe Int64
mJobId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload Maybe Text
mError

-- | The DLQ filters from their query parameters.
dlqFilters :: Maybe Int64 -> Maybe Int64 -> Maybe Text -> Maybe Text -> Maybe Text -> Maybe Text -> [JobFilter]
dlqFilters :: Maybe Int64
-> Maybe Int64
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> [JobFilter]
dlqFilters Maybe Int64
mParentId Maybe Int64
mJobId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload Maybe Text
mError =
  [Maybe JobFilter] -> [JobFilter]
forall a. [Maybe a] -> [a]
catMaybes ([Maybe JobFilter] -> [JobFilter])
-> [Maybe JobFilter] -> [JobFilter]
forall a b. (a -> b) -> a -> b
$
    Maybe Int64
-> Maybe Text -> Maybe Text -> Maybe Text -> [Maybe JobFilter]
listingFilters Maybe Int64
mParentId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload
      [Maybe JobFilter] -> [Maybe JobFilter] -> [Maybe JobFilter]
forall a. Semigroup a => a -> a -> a
<> [Int64 -> JobFilter
FilterJobId (Int64 -> JobFilter) -> Maybe Int64 -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Int64
mJobId, Text -> JobFilter
FilterErrorText (Text -> JobFilter) -> Maybe Text -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text -> Maybe Text
nonBlank Maybe Text
mError]

-- | Run a DLQ retry and report how many trees it restored.
retriedResponse :: ArbiterServerConfig m registry -> (Text -> m Int64) -> Handler BatchRetryResponse
retriedResponse :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> (Text -> m Int64) -> Handler BatchRetryResponse
retriedResponse ArbiterServerConfig m registry
config Text -> m Int64
retry = Int64 -> BatchRetryResponse
BatchRetryResponse (Int64 -> BatchRetryResponse)
-> Handler Int64 -> Handler BatchRetryResponse
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry -> m Int64 -> Handler Int64
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (Text -> m Int64
retry (ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config))

-- | Retry a DLQ job back into the main queue, with its payload replaced when the
-- request carries one. 409 when its parent is gone.
retryFromDLQHandler
  :: forall registry (payload :: Type) m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Maybe (PayloadEdit payload)
  -> Handler NoContent
retryFromDLQHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> Maybe (PayloadEdit payload)
-> Handler NoContent
retryFromDLQHandler Text
tableName ArbiterServerConfig m registry
config Int64
dlqId Maybe (PayloadEdit payload)
edit =
  Either ServerError () -> Handler NoContent
noContentOr (Either ServerError () -> Handler NoContent)
-> Handler (Either ServerError ()) -> Handler NoContent
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< ArbiterServerConfig m registry
-> m (Either ServerError ()) -> Handler (Either ServerError ())
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m (Either ServerError ()) -> m (Either ServerError ())
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction m (Either ServerError ())
retried)
  where
    schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
    retried :: m (Either ServerError ())
retried =
      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 (PayloadEdit payload -> payload
forall payload. PayloadEdit payload -> payload
editPayload (PayloadEdit payload -> payload)
-> Maybe (PayloadEdit payload) -> Maybe payload
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe (PayloadEdit payload)
edit)
        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 (forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
Ops.typedRow @payload)
        m (Maybe (JobRead payload))
-> (Maybe (JobRead payload) -> m (Either ServerError ()))
-> m (Either ServerError ())
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= m (Either ServerError ())
-> (JobRead payload -> m (Either ServerError ()))
-> Maybe (JobRead payload)
-> m (Either ServerError ())
forall b a. b -> (a -> b) -> Maybe a -> b
maybe m (Either ServerError ())
missing (m (Either ServerError ())
-> JobRead payload -> m (Either ServerError ())
forall a b. a -> b -> a
const (Either ServerError () -> m (Either ServerError ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either ServerError ()
forall a b. b -> Either a b
Right ())))
    missing :: m (Either ServerError ())
missing = Bool -> Either ServerError ()
forall {b}. Bool -> Either ServerError b
refuse (Bool -> Either ServerError ())
-> m Bool -> m (Either ServerError ())
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Text -> Int64 -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Bool
Ops.dlqJobExists Text
schemaName Text
tableName Int64
dlqId
    refuse :: Bool -> Either ServerError b
refuse Bool
exists
      | Bool
exists = ServerError -> Either ServerError b
forall a b. a -> Either a b
Left ServerError
err409 {errBody = "Cannot retry: parent job no longer exists (not in queue or DLQ)"}
      | Bool
otherwise = ServerError -> Either ServerError b
forall a b. a -> Either a b
Left ServerError
dlqJobNotFound

-- | Delete a job from DLQ permanently.
deleteDLQHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
deleteDLQHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
deleteDLQHandler Text
tableName ArbiterServerConfig m registry
config Int64
dlqId =
  ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
mutateOr404 ArbiterServerConfig m registry
config ServerError
dlqJobNotFound ((Text -> m Int64) -> Handler NoContent)
-> (Text -> m Int64) -> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.deleteDLQJob Text
schemaName Text
tableName Int64
dlqId

-- | Delete a batch of rows by primary key with this statement.
deleteBatchHandler
  :: ArbiterServerConfig m registry
  -> (Text -> [Int64] -> m Int64)
  -> BatchDeleteRequest
  -> Handler BatchDeleteResponse
deleteBatchHandler :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> (Text -> [Int64] -> m Int64)
-> BatchDeleteRequest
-> Handler BatchDeleteResponse
deleteBatchHandler ArbiterServerConfig m registry
config Text -> [Int64] -> m Int64
deleteRows (BatchDeleteRequest [Int64]
rowIds) = do
  rowsDeleted <- ArbiterServerConfig m registry -> m Int64 -> Handler Int64
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m Int64 -> Handler Int64) -> m Int64 -> Handler Int64
forall a b. (a -> b) -> a -> b
$ Text -> [Int64] -> m Int64
deleteRows (ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config) [Int64]
rowIds
  pure $ BatchDeleteResponse {deleted = rowsDeleted}

-- | Archive API handler for a specific table.
archiveServer
  :: forall registry payload m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> ArchiveAPI payload (AsServerT Handler)
archiveServer :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> ArchiveAPI payload (AsServerT Handler)
archiveServer Text
table ArbiterServerConfig m registry
config =
  ArchiveAPI
    { listArchive :: AsServerT Handler
:- (Summary "List archived jobs"
    :> (Description "The payload filter searches the payload text."
        :> (QueryParam "limit" PageLimit
            :> (QueryParam "offset" Int
                :> (QueryParam "parent_id" Int64
                    :> (QueryParam "job_id" Int64
                        :> (QueryParam "group_key" Text
                            :> (QueryParam "kind" Text
                                :> (QueryParam "payload" Text
                                    :> (QueryParam "completed_after" UTCTime
                                        :> (QueryParam "completed_before" UTCTime
                                            :> (QueryParam "sort_by" ArchiveSortColumn
                                                :> (QueryParam "sort_dir" SortDir
                                                    :> Get
                                                         '[JSON]
                                                         (ArchiveResponse payload))))))))))))))
listArchive = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Int64
-> Maybe Int64
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe UTCTime
-> Maybe UTCTime
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Handler (ArchiveResponse payload)
listArchiveHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , reEnqueueArchive :: AsServerT Handler
:- (Summary "Run an archived job again"
    :> (Description "Inserts a new job. The archive entry stays."
        :> (Capture "id" Int64
            :> ("reenqueue"
                :> (ReqBody'
                      '[Description PayloadEditNote]
                      '[OptionalJSON]
                      (Maybe (PayloadEdit payload))
                    :> PostNoContent)))))
reEnqueueArchive = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> Maybe (PayloadEdit payload)
-> Handler NoContent
reEnqueueArchiveHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , deleteArchive :: AsServerT Handler
:- (Summary "Delete an archived job"
    :> (Capture "id" Int64 :> DeleteNoContent))
deleteArchive = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
deleteArchiveHandler @registry Text
table ArbiterServerConfig m registry
config
    , deleteArchiveBatch :: AsServerT Handler
:- (Summary "Delete archived jobs"
    :> (Description
          "Returns the number deleted. The server skips unknown ids."
        :> ("batch-delete"
            :> (ReqBody '[JSON] BatchDeleteRequest
                :> Post '[JSON] BatchDeleteResponse))))
deleteArchiveBatch = ArbiterServerConfig m registry
-> (Text -> [Int64] -> m Int64)
-> BatchDeleteRequest
-> Handler BatchDeleteResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> (Text -> [Int64] -> m Int64)
-> BatchDeleteRequest
-> Handler BatchDeleteResponse
deleteBatchHandler ArbiterServerConfig m registry
config (Text -> Text -> [Int64] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m Int64
`Ops.deleteArchiveJobsBatch` Text
table)
    }

-- | List archived jobs with pagination and composable filters.
listArchiveHandler
  :: forall registry (payload :: Type) m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Maybe PageLimit
  -> Maybe Int
  -> Maybe Int64
  -> Maybe Int64
  -> Maybe Text
  -> Maybe Text
  -> Maybe Text
  -> Maybe UTCTime
  -> Maybe UTCTime
  -> Maybe ArchiveSortColumn
  -> Maybe SortDir
  -> Handler (ArchiveResponse payload)
listArchiveHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Int64
-> Maybe Int64
-> Maybe Text
-> Maybe Text
-> Maybe Text
-> Maybe UTCTime
-> Maybe UTCTime
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Handler (ArchiveResponse payload)
listArchiveHandler Text
tableName ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset Maybe Int64
mParentId Maybe Int64
mJobId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload Maybe UTCTime
mCompletedAfter Maybe UTCTime
mCompletedBefore Maybe ArchiveSortColumn
mSortBy Maybe SortDir
mSortDir =
  ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> (Int -> Int -> m [ArchiveJob (Stored payload)])
-> m Int64
-> Handler (Page (ArchiveJob (Stored payload)))
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
MonadArbiter m =>
ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> (Int -> Int -> m [a])
-> m Int64
-> Handler (Page a)
readPage ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset Int -> Int -> m [ArchiveJob (Stored payload)]
page (Text -> Text -> [JobFilter] -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [JobFilter] -> m Int64
Ops.countArchiveFiltered Text
schemaName Text
tableName [JobFilter]
filters)
  where
    schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
    page :: Int -> Int -> m [ArchiveJob (Stored payload)]
page = 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
mSortBy Maybe SortDir
mSortDir
    filters :: [JobFilter]
filters =
      [Maybe JobFilter] -> [JobFilter]
forall a. [Maybe a] -> [a]
catMaybes ([Maybe JobFilter] -> [JobFilter])
-> [Maybe JobFilter] -> [JobFilter]
forall a b. (a -> b) -> a -> b
$
        Maybe Int64
-> Maybe Text -> Maybe Text -> Maybe Text -> [Maybe JobFilter]
listingFilters Maybe Int64
mParentId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload
          [Maybe JobFilter] -> [Maybe JobFilter] -> [Maybe JobFilter]
forall a. Semigroup a => a -> a -> a
<> [ Int64 -> JobFilter
FilterJobId (Int64 -> JobFilter) -> Maybe Int64 -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Int64
mJobId
             , UTCTime -> JobFilter
FilterCompletedAfter (UTCTime -> JobFilter) -> Maybe UTCTime -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe UTCTime
mCompletedAfter
             , UTCTime -> JobFilter
FilterCompletedBefore (UTCTime -> JobFilter) -> Maybe UTCTime -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe UTCTime
mCompletedBefore
             ]

-- | Re-enqueue an archived job as a fresh job, with its payload replaced when the
-- request carries one. 404 if the archive row is gone.
reEnqueueArchiveHandler
  :: forall registry (payload :: Type) m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Maybe (PayloadEdit payload)
  -> Handler NoContent
reEnqueueArchiveHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> Maybe (PayloadEdit payload)
-> Handler NoContent
reEnqueueArchiveHandler Text
tableName ArbiterServerConfig m registry
config Int64
archiveId Maybe (PayloadEdit payload)
edit = do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
      reEnqueue :: m (Maybe (JobRead (Stored payload)))
reEnqueue = 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 (PayloadEdit payload -> payload
forall payload. PayloadEdit payload -> payload
editPayload (PayloadEdit payload -> payload)
-> Maybe (PayloadEdit payload) -> Maybe payload
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe (PayloadEdit payload)
edit)
  NoContent
NoContent
    NoContent -> Handler (JobRead payload) -> Handler NoContent
forall a b. a -> Handler b -> Handler a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ ArbiterServerConfig m registry
-> m (Maybe (JobRead payload))
-> ServerError
-> Handler (JobRead payload)
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> m (Maybe a) -> ServerError -> Handler a
orNotFound ArbiterServerConfig m registry
config (m (Maybe (JobRead payload)) -> m (Maybe (JobRead payload))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Maybe (JobRead (Stored payload)))
reEnqueue 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 (forall payload (m :: * -> *).
(FromJSON payload, MonadArbiter m) =>
JobRead (Stored payload) -> m (JobRead payload)
Ops.typedRow @payload))) ServerError
archivedJobNotFound

-- | Purge one archived job by its archive primary key.
deleteArchiveHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> Handler NoContent
deleteArchiveHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry -> Int64 -> Handler NoContent
deleteArchiveHandler Text
tableName ArbiterServerConfig m registry
config Int64
archiveId =
  ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
mutateOr404 ArbiterServerConfig m registry
config ServerError
archivedJobNotFound ((Text -> m Int64) -> Handler NoContent)
-> (Text -> m Int64) -> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName -> Text -> Text -> Int64 -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m Int64
Ops.deleteArchiveJob Text
schemaName Text
tableName Int64
archiveId

-- | Stats API handler for a specific table.
statsServer
  :: forall registry payload m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> StatsAPI (AsServerT Handler)
statsServer :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry -> StatsAPI (AsServerT Handler)
statsServer Text
tableName ArbiterServerConfig m registry
config =
  StatsAPI
    { getStats :: AsServerT Handler
:- (Summary "Show queue stats" :> Get '[JSON] StatsResponse)
getStats = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> [Text]
-> Handler StatsResponse
getStatsHandler @registry Text
tableName ArbiterServerConfig m registry
config (forall payload. HasKind payload => [Text]
kindsFor @payload)
    }

-- | Get queue statistics.
getStatsHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> [Text]
  -> Handler StatsResponse
getStatsHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> [Text]
-> Handler StatsResponse
getStatsHandler Text
tableName ArbiterServerConfig m registry
config [Text]
kinds =
  IO StatsResponse -> Handler StatsResponse
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO StatsResponse -> Handler StatsResponse)
-> IO StatsResponse -> Handler StatsResponse
forall a b. (a -> b) -> a -> b
$ NominalDiffTime
-> CacheCell StatsResponse
-> Text
-> IO StatsResponse
-> IO StatsResponse
forall a. NominalDiffTime -> CacheCell a -> Text -> IO a -> IO a
cachedForKey (ArbiterServerConfig m registry -> NominalDiffTime
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
statsCacheTtl ArbiterServerConfig m registry
config) (ArbiterServerConfig m registry -> CacheCell StatsResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell StatsResponse
queueStatsCache ArbiterServerConfig m registry
config) Text
tableName (IO StatsResponse -> IO StatsResponse)
-> IO StatsResponse -> IO StatsResponse
forall a b. (a -> b) -> a -> b
$ do
    let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config

    queueStats <- ArbiterServerConfig m registry -> m QueueStats -> IO QueueStats
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m QueueStats -> IO QueueStats) -> m QueueStats -> IO QueueStats
forall a b. (a -> b) -> a -> b
$ Text -> Text -> [Text] -> m QueueStats
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Text] -> m QueueStats
Ops.getQueueStats Text
schemaName Text
tableName [Text]
kinds
    now <- getCurrentTime

    pure $ StatsResponse {stats = queueStats, timestamp = now}

-- | Every queue's stats in one request, for the landing overview.
getAllStatsHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> [(Text, [Text])]
  -> Handler AllStatsResponse
getAllStatsHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> [(Text, [Text])] -> Handler AllStatsResponse
getAllStatsHandler ArbiterServerConfig m registry
config [(Text, [Text])]
queueKinds =
  ArbiterServerConfig m registry
-> CacheCell AllStatsResponse
-> ([QueueOverview] -> AllStatsResponse)
-> m [QueueOverview]
-> Handler AllStatsResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry) a b.
ArbiterServerConfig m registry
-> CacheCell a -> (b -> a) -> m b -> Handler a
cachedRead ArbiterServerConfig m registry
config (ArbiterServerConfig m registry -> CacheCell AllStatsResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell AllStatsResponse
allQueueStatsCache ArbiterServerConfig m registry
config) (\[QueueOverview]
overviews -> AllStatsResponse {queues :: [QueueOverview]
queues = [QueueOverview]
overviews}) (m [QueueOverview] -> Handler AllStatsResponse)
-> m [QueueOverview] -> Handler AllStatsResponse
forall a b. (a -> b) -> a -> b
$
    Text -> [(Text, [Text])] -> m [QueueOverview]
forall (m :: * -> *).
MonadArbiter m =>
Text -> [(Text, [Text])] -> m [QueueOverview]
Ops.getAllQueueStats (ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config) [(Text, [Text])]
queueKinds

-- | Table API handlers for a specific table.
tableServer
  :: forall registry payload result m
   . (EncodeJobResult result, HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> TableAPI payload result (AsServerT Handler)
tableServer :: forall (registry :: JobPayloadRegistry) payload result
       (m :: * -> *).
(EncodeJobResult result, HasRegistry m registry,
 JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> TableAPI payload result (AsServerT Handler)
tableServer Text
table ArbiterServerConfig m registry
config =
  TableAPI
    { jobs :: AsServerT Handler
:- ("jobs" :> NamedRoutes (JobsAPI payload result))
jobs = forall (registry :: JobPayloadRegistry) payload result
       (m :: * -> *).
(EncodeJobResult result, HasRegistry m registry,
 JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> JobsAPI payload result (AsServerT Handler)
jobsServer @registry @payload @result Text
table ArbiterServerConfig m registry
config
    , claimJobs :: AsServerT Handler
:- (Summary "Claim jobs"
    :> (Description
          "Leases up to maxJobs visible jobs for leaseSeconds. Each job carries the claimSeq and claimedBy that ack, nack and extend need. The server does not renew the lease. A paused queue returns no jobs."
        :> ("claim"
            :> (ReqBody '[JSON] ClaimRequest
                :> Post '[JSON] (ClaimResponse payload)))))
claimJobs = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> ClaimRequest
-> Handler (ClaimResponse payload)
claimJobsHandler @registry @payload Text
table ArbiterServerConfig m registry
config
    , dlq :: AsServerT Handler :- ("dlq" :> NamedRoutes (DLQAPI payload))
dlq = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> DLQAPI payload (AsServerT Handler)
dlqServer @registry @payload Text
table ArbiterServerConfig m registry
config
    , archive :: AsServerT Handler
:- ("archive" :> NamedRoutes (ArchiveAPI payload))
archive = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> ArchiveAPI payload (AsServerT Handler)
archiveServer @registry @payload Text
table ArbiterServerConfig m registry
config
    , stats :: AsServerT Handler :- ("stats" :> NamedRoutes StatsAPI)
stats = forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry -> StatsAPI (AsServerT Handler)
statsServer @registry @payload Text
table ArbiterServerConfig m registry
config
    , listKinds :: AsServerT Handler
:- (Summary "List payload kinds"
    :> ("kinds" :> Get '[JSON] [Text]))
listKinds = [Text] -> Handler [Text]
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (forall payload. HasKind payload => [Text]
kindsFor @payload)
    , listGroups :: AsServerT Handler
:- (Summary "List open groups"
    :> (Description "The largest groups come first."
        :> ("groups"
            :> (QueryParam "limit" PageLimit
                :> (QueryParam "offset" Int
                    :> (QueryParam "group_key" Text :> Get '[JSON] GroupsResponse))))))
listGroups = forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Text
-> Handler GroupsResponse
listGroupsHandler @registry Text
table ArbiterServerConfig m registry
config
    , getDetails :: AsServerT Handler
:- (Summary "Show a queue's pause state and DLQ retention"
    :> (Description
          "Null when the queue has no row. Migrations create one for each queue."
        :> ("details" :> Get '[JSON] (Maybe QueueRow))))
getDetails = ArbiterServerConfig m registry
-> m (Maybe QueueRow) -> Handler (Maybe QueueRow)
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (Text -> Text -> m (Maybe QueueRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe QueueRow)
Ops.getQueue (ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config) Text
table)
    , pauseQueue :: AsServerT Handler
:- (Summary "Pause a queue"
    :> (Description
          "Workers do not claim from the queue. Claims through this API return no jobs."
        :> ("pause" :> PostNoContent)))
pauseQueue = ArbiterServerConfig m registry -> Text -> Bool -> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Text -> Bool -> Handler NoContent
setQueuePausedHandler ArbiterServerConfig m registry
config Text
table Bool
True
    , resumeQueue :: AsServerT Handler
:- (Summary "Resume a queue" :> ("resume" :> PostNoContent))
resumeQueue = ArbiterServerConfig m registry -> Text -> Bool -> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Text -> Bool -> Handler NoContent
setQueuePausedHandler ArbiterServerConfig m registry
config Text
table Bool
False
    , setDLQRetention :: AsServerT Handler
:- (Summary "Set a queue's DLQ retention"
    :> (Description
          "The reaper deletes the queue's DLQ jobs that failed more than retentionSeconds ago. A null retentionSeconds keeps them. Every worker pool sharing the schema applies it."
        :> (Throws 400 "retentionSeconds is not positive."
            :> ("dlq-retention"
                :> (ReqBody '[JSON] DLQRetentionRequest :> PutNoContent)))))
setDLQRetention = ArbiterServerConfig m registry
-> Text -> DLQRetentionRequest -> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text -> DLQRetentionRequest -> Handler NoContent
setDLQRetentionHandler ArbiterServerConfig m registry
config Text
table
    }

-- | A page of a queue's open groups, largest first.
listGroupsHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Maybe PageLimit
  -> Maybe Int
  -> Maybe Text
  -> Handler GroupsResponse
listGroupsHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> Maybe Text
-> Handler GroupsResponse
listGroupsHandler Text
tableName ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset Maybe Text
mGroupKey =
  ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> (Int -> Int -> m [GroupSummary])
-> m Int64
-> Handler GroupsResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
MonadArbiter m =>
ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> (Int -> Int -> m [a])
-> m Int64
-> Handler (Page a)
readPage ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset (Text -> Text -> Maybe Text -> Int -> Int -> m [GroupSummary]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe Text -> Int -> Int -> m [GroupSummary]
Ops.listGroups Text
schemaName Text
tableName Maybe Text
key) (Text -> Text -> Maybe Text -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe Text -> m Int64
Ops.countGroups Text
schemaName Text
tableName Maybe Text
key)
  where
    schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
    key :: Maybe Text
key = Maybe Text -> Maybe Text
nonBlank Maybe Text
mGroupKey

-- | Read one page and the total it pages through in one transaction.
readPage
  :: (MonadArbiter m)
  => ArbiterServerConfig m registry
  -> Maybe PageLimit
  -> Maybe Int
  -> (Int -> Int -> m [a])
  -> m Int64
  -> Handler (Page a)
readPage :: forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
MonadArbiter m =>
ArbiterServerConfig m registry
-> Maybe PageLimit
-> Maybe Int
-> (Int -> Int -> m [a])
-> m Int64
-> Handler (Page a)
readPage ArbiterServerConfig m registry
config Maybe PageLimit
mLimit Maybe Int
mOffset Int -> Int -> m [a]
page m Int64
count = do
  let pagination :: (Int, Int)
pagination@(Int
limit, Int
offset) = Maybe PageLimit -> Maybe Int -> (Int, Int)
forall (def :: Nat).
KnownNat def =>
Maybe (PageSize def) -> Maybe Int -> (Int, Int)
validatePagination Maybe PageLimit
mLimit Maybe Int
mOffset
  (rows, total) <- ArbiterServerConfig m registry
-> m ([a], Int64) -> Handler ([a], Int64)
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m ([a], Int64) -> Handler ([a], Int64))
-> (m ([a], Int64) -> m ([a], Int64))
-> m ([a], Int64)
-> Handler ([a], Int64)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. m ([a], Int64) -> m ([a], Int64)
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m ([a], Int64) -> Handler ([a], Int64))
-> m ([a], Int64) -> Handler ([a], Int64)
forall a b. (a -> b) -> a -> b
$ (,) ([a] -> Int64 -> ([a], Int64))
-> m [a] -> m (Int64 -> ([a], Int64))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Int -> Int -> m [a]
page Int
limit Int
offset m (Int64 -> ([a], Int64)) -> m Int64 -> m ([a], Int64)
forall a b. m (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> m Int64
count
  pure (toPage pagination total rows)

-- | Rows read at a limit and offset, and the total they page through.
toPage :: (Int, Int) -> Int64 -> [a] -> Page a
toPage :: forall a. (Int, Int) -> Int64 -> [a] -> Page a
toPage (Int
limit, Int
offset) Int64
total [a]
rows =
  Page {pageItems :: [a]
pageItems = [a]
rows, pageTotal :: Int
pageTotal = Int64 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int64
total, pageOffset :: Int
pageOffset = Int
offset, pageLimit :: Int
pageLimit = Int
limit}

-- | Lease visible jobs to a consumer outside a worker pool. Each returned job
-- contains the claim sequence and claimant required for finalization.
claimJobsHandler
  :: forall registry payload m
   . (HasRegistry m registry, JobPayload payload)
  => Text
  -> ArbiterServerConfig m registry
  -> ClaimRequest
  -> Handler (ClaimResponse payload)
claimJobsHandler :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
(HasRegistry m registry, JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> ClaimRequest
-> Handler (ClaimResponse payload)
claimJobsHandler Text
tableName ArbiterServerConfig m registry
config ClaimRequest
req = IO (ClaimResponse payload) -> Handler (ClaimResponse payload)
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ClaimResponse payload) -> Handler (ClaimResponse payload))
-> IO (ClaimResponse payload) -> Handler (ClaimResponse payload)
forall a b. (a -> b) -> a -> b
$ do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
      wanted :: Int
wanted = (Int, Int) -> Int -> Int
forall a. Ord a => (a, a) -> a -> a
clamp (Int, Int)
claimJobsRange (Int -> Maybe Int -> Int
forall a. a -> Maybe a -> a
fromMaybe Int
defaultClaimJobs (ClaimRequest -> Maybe Int
maxJobs ClaimRequest
req))
      leaseSecs :: NominalDiffTime
leaseSecs = Double -> NominalDiffTime
forall a b. (Real a, Fractional b) => a -> b
realToFrac ((Double, Double) -> Double -> Double
forall a. Ord a => (a, a) -> a -> a
clamp (Double, Double)
leaseSecondsRange (Double -> Maybe Double -> Double
forall a. a -> Maybe a -> a
fromMaybe Double
defaultLeaseSeconds (ClaimRequest -> Maybe Double
leaseSeconds ClaimRequest
req)))
  claimant <- IO UUID
UUID.nextRandom
  claimed <- runDb config $ do
    mQueue <- Ops.getQueue schemaName tableName
    if any Queues.paused mQueue
      then pure []
      else do
        (jobs, rejected) <-
          Ops.claimJobsCached
            (Ops.mkJobStatements @payload schemaName tableName Ops.singleJobBatchSize Ops.noStagedCapacities leaseSecs claimant)
            wanted
        traverse_ deadLetter rejected
        pure jobs
  pure ClaimResponse {jobs = claimed}
  where
    deadLetter :: RejectedRow payload -> m (Either SomeException Int64)
deadLetter RejectedRow payload
rejected =
      LogConfig
-> LogLevel
-> FailureGates
-> Text
-> m Int64
-> m (Either SomeException Int64)
forall (m :: * -> *) a.
MonadUnliftIO m =>
LogConfig
-> LogLevel
-> FailureGates
-> Text
-> m a
-> m (Either SomeException a)
tryReportedOn
        (ArbiterServerConfig m registry -> LogConfig
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> LogConfig
serverLogConfig ArbiterServerConfig m registry
config)
        LogLevel
Error
        (ArbiterServerConfig m registry -> FailureGates
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> FailureGates
deadLetterGates ArbiterServerConfig m registry
config)
        (Text
"Dead-letter undecodable job in " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName)
        (RejectedRow payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
RejectedRow payload -> m Int64
Ops.deadLetterRejected RejectedRow payload
rejected)

-- | Complete a job that the caller holds. Store an optional result in the
-- parent rollup or archive entry, as worker @ackWith@ does.
ackClaimedJobHandler
  :: forall registry result m
   . (EncodeJobResult result, HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> AckRequest result
  -> Handler NoContent
ackClaimedJobHandler :: forall (registry :: JobPayloadRegistry) result (m :: * -> *).
(EncodeJobResult result, HasRegistry m registry) =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> AckRequest result
-> Handler NoContent
ackClaimedJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId AckRequest result
req =
  forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> JobLease
-> (Text -> JobRead (Stored payload) -> m Int64)
-> Handler NoContent
withHeldJob @registry Text
tableName ArbiterServerConfig m registry
config Int64
jobId (AckRequest result -> JobLease
forall result. AckRequest result -> JobLease
arLease AckRequest result
req) ((Text -> JobRead (Stored (ZonkAny 5)) -> m Int64)
 -> Handler NoContent)
-> (Text -> JobRead (Stored (ZonkAny 5)) -> m Int64)
-> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName JobRead (Stored (ZonkAny 5))
job ->
    m Int64 -> m Int64
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m Int64 -> m Int64) -> m Int64 -> m Int64
forall a b. (a -> b) -> a -> b
$ do
      rows <- Text -> Text -> JobRead (Stored (ZonkAny 5)) -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> JobRead payload -> m Int64
Ops.ackJob Text
schemaName Text
tableName JobRead (Stored (ZonkAny 5))
job
      when (rows > 0) $ storeEncodedResult job (arResult req >>= encodeJobResult)
      pure rows

-- | Restore the attempt used by a claim. The job becomes available when its
-- lease expires.
nackClaimedJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> JobLease
  -> Handler NoContent
nackClaimedJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> JobLease
-> Handler NoContent
nackClaimedJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId JobLease
lease =
  forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> JobLease
-> (Text -> JobRead (Stored payload) -> m Int64)
-> Handler NoContent
withHeldJob @registry Text
tableName ArbiterServerConfig m registry
config Int64
jobId JobLease
lease ((Text -> JobRead (Stored (ZonkAny 6)) -> m Int64)
 -> Handler NoContent)
-> (Text -> JobRead (Stored (ZonkAny 6)) -> m Int64)
-> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName JobRead (Stored (ZonkAny 6))
job ->
    Text
-> Text
-> Maybe NominalDiffTime
-> JobRead (Stored (ZonkAny 6))
-> 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 (Stored (ZonkAny 6))
job

-- | Extend a held HTTP lease, equivalent to a worker heartbeat.
extendClaimedJobHandler
  :: forall registry m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> ExtendRequest
  -> Handler NoContent
extendClaimedJobHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> ExtendRequest
-> Handler NoContent
extendClaimedJobHandler Text
tableName ArbiterServerConfig m registry
config Int64
jobId ExtendRequest
req =
  forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> JobLease
-> (Text -> JobRead (Stored payload) -> m Int64)
-> Handler NoContent
withHeldJob @registry Text
tableName ArbiterServerConfig m registry
config Int64
jobId (ExtendRequest -> JobLease
erLease ExtendRequest
req) ((Text -> JobRead (Stored (ZonkAny 7)) -> m Int64)
 -> Handler NoContent)
-> (Text -> JobRead (Stored (ZonkAny 7)) -> m Int64)
-> Handler NoContent
forall a b. (a -> b) -> a -> b
$ \Text
schemaName JobRead (Stored (ZonkAny 7))
job ->
    Text
-> Text
-> NominalDiffTime
-> JobRead (Stored (ZonkAny 7))
-> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> NominalDiffTime -> JobRead payload -> m Int64
Ops.setVisibilityTimeout Text
schemaName Text
tableName (Double -> NominalDiffTime
forall a b. (Real a, Fractional b) => a -> b
realToFrac ((Double, Double) -> Double -> Double
forall a. Ord a => (a, a) -> a -> a
clamp (Double, Double)
leaseSecondsRange (ExtendRequest -> Double
erSeconds ExtendRequest
req))) JobRead (Stored (ZonkAny 7))
job

-- | Finalize the job identified by a lease. Refuse a lease that the caller no
-- longer holds or a lease held by a registered worker. Each statement checks the claim
-- sequence and writes no change after a lease is lost.
withHeldJob
  :: forall registry (payload :: Type) m
   . (HasRegistry m registry)
  => Text
  -> ArbiterServerConfig m registry
  -> Int64
  -> JobLease
  -> (Text -> Job.JobRead (Job.Stored payload) -> m Int64)
  -> Handler NoContent
withHeldJob :: forall (registry :: JobPayloadRegistry) payload (m :: * -> *).
HasRegistry m registry =>
Text
-> ArbiterServerConfig m registry
-> Int64
-> JobLease
-> (Text -> JobRead (Stored payload) -> m Int64)
-> Handler NoContent
withHeldJob Text
tableName ArbiterServerConfig m registry
config Int64
jobId JobLease
lease Text -> JobRead (Stored payload) -> m Int64
finalize =
  Either ServerError () -> Handler NoContent
noContentOr
    (Either ServerError () -> Handler NoContent)
-> Handler (Either ServerError ()) -> Handler NoContent
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< ArbiterServerConfig m registry
-> m (Either ServerError ()) -> Handler (Either ServerError ())
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> Int64 -> m (Maybe (JobRead (Stored payload)))
Ops.getJobById @_ @payload Text
schemaName Text
tableName Int64
jobId m (Maybe (JobRead (Stored payload)))
-> (Maybe (JobRead (Stored payload)) -> m (Either ServerError ()))
-> m (Either ServerError ())
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= m (Either ServerError ())
-> (JobRead (Stored payload) -> m (Either ServerError ()))
-> Maybe (JobRead (Stored payload))
-> m (Either ServerError ())
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (Either ServerError () -> m (Either ServerError ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Either ServerError ()
forall {b}. Either ServerError b
notFound) JobRead (Stored payload) -> m (Either ServerError ())
finalizeHeld)
  where
    schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
    notFound :: Either ServerError b
notFound = ServerError -> Either ServerError b
forall a b. a -> Either a b
Left ServerError
jobNotFound
    refuse :: ByteString -> Either ServerError b
refuse ByteString
body = ServerError -> Either ServerError b
forall a b. a -> Either a b
Left ServerError
err409 {errBody = body}
    held :: JobRead (Stored payload) -> Bool
held JobRead (Stored payload)
job = JobRead (Stored payload) -> Maybe UUID
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe UUID
Job.claimedBy JobRead (Stored payload)
job Maybe UUID -> Maybe UUID -> Bool
forall a. Eq a => a -> a -> Bool
== UUID -> Maybe UUID
forall a. a -> Maybe a
Just (JobLease -> UUID
jlClaimedBy JobLease
lease) Bool -> Bool -> Bool
&& JobRead (Stored payload) -> Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int64
Job.claimSeq JobRead (Stored payload)
job Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
== JobLease -> Int64
jlClaimSeq JobLease
lease
    finalizeHeld :: JobRead (Stored payload) -> m (Either ServerError ())
finalizeHeld JobRead (Stored payload)
job
      | Bool -> Bool
not (JobRead (Stored payload) -> Bool
held JobRead (Stored payload)
job) = Either ServerError () -> m (Either ServerError ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> Either ServerError ()
forall {b}. ByteString -> Either ServerError b
refuse ByteString
"Job is not held by this lease")
      | Bool
otherwise = do
          pooled <- Text -> UUID -> m Bool
forall (m :: * -> *). MonadArbiter m => Text -> UUID -> m Bool
Ops.workerRegistered Text
schemaName (JobLease -> UUID
jlClaimedBy JobLease
lease)
          if pooled
            then pure (refuse "Job is held by a registered worker")
            else decide job <$> finalize schemaName job
    decide :: Job payload Int64 q insertedAt adm -> a -> Either ServerError ()
decide Job payload Int64 q insertedAt adm
job a
rowsAffected
      | a
rowsAffected a -> a -> Bool
forall a. Ord a => a -> a -> Bool
> a
0 = () -> Either ServerError ()
forall a b. b -> Either a b
Right ()
      | Job payload Int64 q insertedAt adm -> Bool
forall p q t adm. Job p Int64 q t adm -> Bool
Job.suspended Job payload Int64 q insertedAt adm
job = ByteString -> Either ServerError ()
forall {b}. ByteString -> Either ServerError b
refuse ByteString
"Job is suspended"
      | Bool
otherwise = ByteString -> Either ServerError ()
forall {b}. ByteString -> Either ServerError b
refuse ByteString
"Lease no longer held"

-- | Maintenance API handler.
maintenanceServer
  :: forall registry m
   . (HL.RegistryAdmissionPolicies registry, HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> MaintenanceAPI (AsServerT Handler)
maintenanceServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(RegistryAdmissionPolicies registry, HasRegistry m registry,
 RegistryTables registry) =>
ArbiterServerConfig m registry
-> MaintenanceAPI (AsServerT Handler)
maintenanceServer ArbiterServerConfig m registry
config = MaintenanceAPI {runMaintenance :: AsServerT Handler
:- (Summary "Run a maintenance pass"
    :> (Description
          "Runs the work of the reaper once. An operation that another caller runs at the same time is skipped and is not in the response."
        :> Post '[JSON] MaintenanceResponse))
runMaintenance = forall (registry :: JobPayloadRegistry) (m :: * -> *).
(RegistryAdmissionPolicies registry, HasRegistry m registry,
 RegistryTables registry) =>
ArbiterServerConfig m registry -> Handler MaintenanceResponse
maintenanceHandler @registry ArbiterServerConfig m registry
config}

-- | Run one maintenance pass. A worker pool's reaper does the same work. Operations
-- exclude each other across callers. An operation another caller is running is
-- skipped and absent from the response.
maintenanceHandler
  :: forall registry m
   . (HL.RegistryAdmissionPolicies registry, HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> Handler MaintenanceResponse
maintenanceHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(RegistryAdmissionPolicies registry, HasRegistry m registry,
 RegistryTables registry) =>
ArbiterServerConfig m registry -> Handler MaintenanceResponse
maintenanceHandler ArbiterServerConfig m registry
config = IO MaintenanceResponse -> Handler MaintenanceResponse
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO MaintenanceResponse -> Handler MaintenanceResponse)
-> IO MaintenanceResponse -> Handler MaintenanceResponse
forall a b. (a -> b) -> a -> b
$ do
  touched <- Map Text Int64 -> IO (IORef (Map Text Int64))
forall a. a -> IO (IORef a)
newIORef Map Text Int64
forall k a. Map k a
Map.empty
  let report MaintenanceOp
operation Int64
rows = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ IORef (Map Text Int64)
-> (Map Text Int64 -> Map Text Int64) -> IO ()
forall a. IORef a -> (a -> a) -> IO ()
modifyIORef' IORef (Map Text Int64)
touched ((Int64 -> Int64 -> Int64)
-> Text -> Int64 -> Map Text Int64 -> Map Text Int64
forall k a. Ord k => (a -> a -> a) -> k -> a -> Map k a -> Map k a
Map.insertWith Int64 -> Int64 -> Int64
forall a. Num a => a -> a -> a
(+) (MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
operation) Int64
rows)
      pace =
        MaintenancePace
          { paceWindow :: NominalDiffTime
paceWindow = ArbiterServerConfig m registry -> NominalDiffTime
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceInterval ArbiterServerConfig m registry
config
          , paceSparseWindow :: NominalDiffTime
paceSparseWindow = ArbiterServerConfig m registry -> NominalDiffTime
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceSparseInterval ArbiterServerConfig m registry
config
          , paceBucketIdle :: NominalDiffTime
paceBucketIdle = ArbiterServerConfig m registry -> NominalDiffTime
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceBucketIdle ArbiterServerConfig m registry
config
          }
  failed <-
    runDb config $
      runMaintenancePass (serverLogConfig config) report pace (maintenanceTimeout config)
  ops <- readIORef touched
  pure MaintenanceResponse {maintenanceOps = ops, maintenanceFailed = map maintenanceOpName failed}

-- | Queues API handler.
queuesServer
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => Proxy registry
  -> ArbiterServerConfig m registry
  -> QueuesAPI (AsServerT Handler)
queuesServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
Proxy registry
-> ArbiterServerConfig m registry -> QueuesAPI (AsServerT Handler)
queuesServer Proxy registry
registryProxy ArbiterServerConfig m registry
config =
  QueuesAPI
    { listQueues :: AsServerT Handler
:- (Summary "List queues" :> Get '[JSON] QueuesResponse)
listQueues = QueuesResponse -> Handler QueuesResponse
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (QueuesResponse -> Handler QueuesResponse)
-> QueuesResponse -> Handler QueuesResponse
forall a b. (a -> b) -> a -> b
$ QueuesResponse {queues :: [Text]
queues = Proxy registry -> [Text]
forall (registry :: JobPayloadRegistry).
RegistryTables registry =>
Proxy registry -> [Text]
registryTableNames Proxy registry
registryProxy}
    , getAllStats :: AsServerT Handler
:- (Summary "Show stats for every queue"
    :> ("stats" :> Get '[JSON] AllStatsResponse))
getAllStats = ArbiterServerConfig m registry
-> [(Text, [Text])] -> Handler AllStatsResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> [(Text, [Text])] -> Handler AllStatsResponse
getAllStatsHandler ArbiterServerConfig m registry
config (Proxy registry -> [(Text, [Text])]
forall (registry :: JobPayloadRegistry).
RegistryTables registry =>
Proxy registry -> [(Text, [Text])]
registryQueueKinds Proxy registry
registryProxy)
    }

-- | Set a queue's DLQ retention. 400 for a retention that is not positive.
setDLQRetentionHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Text
  -> DLQRetentionRequest
  -> Handler NoContent
setDLQRetentionHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text -> DLQRetentionRequest -> Handler NoContent
setDLQRetentionHandler ArbiterServerConfig m registry
config Text
queue (DLQRetentionRequest Maybe Double
retention) = do
  Bool -> Handler () -> Handler ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ((Double -> Bool) -> Maybe Double -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
<= Double
0) Maybe Double
retention) (Handler () -> Handler ()) -> Handler () -> Handler ()
forall a b. (a -> b) -> a -> b
$ ServerError -> Handler ()
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err400 {errBody = "retentionSeconds must be positive"}
  NoContent
NoContent NoContent -> Handler () -> Handler NoContent
forall a b. a -> Handler b -> Handler a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ ArbiterServerConfig m registry -> m () -> Handler ()
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (Text -> Text -> Maybe NominalDiffTime -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Maybe NominalDiffTime -> m ()
Ops.setQueueDLQRetention (ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config) Text
queue (Double -> NominalDiffTime
forall a b. (Real a, Fractional b) => a -> b
realToFrac (Double -> NominalDiffTime)
-> Maybe Double -> Maybe NominalDiffTime
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Double
retention))

-- | Flip the @paused@ flag for a queue, creating its @arbiter_queues@ row when missing.
setQueuePausedHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Text
  -> Bool
  -> Handler NoContent
setQueuePausedHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Text -> Bool -> Handler NoContent
setQueuePausedHandler ArbiterServerConfig m registry
config Text
queue Bool
pauseFlag =
  -- The landing overview shows each queue's paused flag.
  CacheCell StatsResponse -> Handler NoContent -> Handler NoContent
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry -> CacheCell StatsResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell StatsResponse
queueStatsCache ArbiterServerConfig m registry
config) (Handler NoContent -> Handler NoContent)
-> (Handler NoContent -> Handler NoContent)
-> Handler NoContent
-> Handler NoContent
forall b c a. (b -> c) -> (a -> b) -> a -> c
. CacheCell AllStatsResponse
-> Handler NoContent -> Handler NoContent
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry -> CacheCell AllStatsResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell AllStatsResponse
allQueueStatsCache ArbiterServerConfig m registry
config) (Handler NoContent -> Handler NoContent)
-> Handler NoContent -> Handler NoContent
forall a b. (a -> b) -> a -> b
$
    NoContent
NoContent NoContent -> Handler Int64 -> Handler NoContent
forall a b. a -> Handler b -> Handler a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ ArbiterServerConfig m registry -> m Int64 -> Handler Int64
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (Text -> Text -> Bool -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Bool -> m Int64
Ops.setQueuePaused (ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config) Text
queue Bool
pauseFlag)

-- | Serve the SSE stream as a raw WAI application. Each client registers on the
-- backend's shared listener for the response's lifetime and gets a @connected@
-- event once its channel is subscribed. If 'enableSSE' is false or the backend
-- has no listener, send one @disabled@ event and close the stream.
eventsServer
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Tagged Handler Application
eventsServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Tagged Handler Application
eventsServer ArbiterServerConfig m registry
config = Application -> Tagged Handler Application
forall {k} (s :: k) b. b -> Tagged s b
Tagged (Application -> Tagged Handler Application)
-> Application -> Tagged Handler Application
forall a b. (a -> b) -> a -> b
$ \Request
_req Response -> IO ResponseReceived
sendResponse -> do
  mListener <- if ArbiterServerConfig m registry -> Bool
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Bool
enableSSE ArbiterServerConfig m registry
config then ArbiterServerConfig m registry
-> m (Maybe Listener) -> IO (Maybe Listener)
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config m (Maybe Listener)
forall (m :: * -> *). MonadArbiter m => m (Maybe Listener)
getListener else Maybe Listener -> IO (Maybe Listener)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe Listener
forall a. Maybe a
Nothing
  case mListener of
    Maybe Listener
Nothing -> Response -> IO ResponseReceived
sendResponse (Response -> IO ResponseReceived)
-> Response -> IO ResponseReceived
forall a b. (a -> b) -> a -> b
$ Status -> ResponseHeaders -> StreamingBody -> Response
responseStream Status
status200 ResponseHeaders
sseHeaders (StreamingBody -> Response) -> StreamingBody -> Response
forall a b. (a -> b) -> a -> b
$ \Builder -> IO ()
write IO ()
flush -> do
      Builder -> IO ()
write Builder
"data: {\"event\":\"disabled\"}\n\n"
      IO ()
flush
    Just Listener
listener -> do
      events <- IO (TChan StrictByteString)
forall a. IO (TChan a)
newTChanIO
      let deliver = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ())
-> (Notification -> STM ()) -> Notification -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TChan StrictByteString -> StrictByteString -> STM ()
forall a. TChan a -> a -> STM ()
writeTChan TChan StrictByteString
events (StrictByteString -> STM ())
-> (Notification -> StrictByteString) -> Notification -> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Notification -> StrictByteString
notificationData
      -- The registration wraps the whole response. The hub is released when the
      -- streaming body never runs.
      withChannels listener (toHubLog (serverLogConfig config)) [(eventStreamingChannel, deliver)] $ \STM Bool
ready ->
        Response -> IO ResponseReceived
sendResponse (Response -> IO ResponseReceived)
-> Response -> IO ResponseReceived
forall a b. (a -> b) -> a -> b
$ Status -> ResponseHeaders -> StreamingBody -> Response
responseStream Status
status200 ResponseHeaders
sseHeaders (StreamingBody -> Response) -> StreamingBody -> Response
forall a b. (a -> b) -> a -> b
$ \Builder -> IO ()
write IO ()
flush ->
          -- A failed write (client gone) ends the stream. The keepalive comment
          -- every 15s is how a gone client is noticed.
          (SomeException -> IO ()) -> IO () -> IO ()
forall (m :: * -> *) a.
MonadUnliftIO m =>
(SomeException -> m a) -> m a -> m a
handleAny (IO () -> SomeException -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
            let keepalive :: IO ()
keepalive = Builder -> IO ()
write Builder
": keepalive\n\n" IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
flush
                event :: StrictByteString -> IO ()
event StrictByteString
payload = Builder -> IO ()
write (Builder
"data: " Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> StrictByteString -> Builder
Builder.byteString StrictByteString
payload Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> Builder
"\n\n") IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
flush
                awaitReady :: IO ()
awaitReady = Int -> IO () -> IO (Maybe ())
forall a. Int -> IO a -> IO (Maybe a)
timeout Int
sseKeepaliveMicros (STM () -> IO ()
forall a. STM a -> IO a
atomically (STM Bool
ready STM Bool -> (Bool -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM ()
check)) IO (Maybe ()) -> (Maybe () -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IO () -> (() -> IO ()) -> Maybe () -> IO ()
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (IO ()
keepalive IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
awaitReady) () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
                pump :: IO ()
pump = Int -> IO StrictByteString -> IO (Maybe StrictByteString)
forall a. Int -> IO a -> IO (Maybe a)
timeout Int
sseKeepaliveMicros (STM StrictByteString -> IO StrictByteString
forall a. STM a -> IO a
atomically (TChan StrictByteString -> STM StrictByteString
forall a. TChan a -> STM a
readTChan TChan StrictByteString
events)) IO (Maybe StrictByteString)
-> (Maybe StrictByteString -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IO ()
-> (StrictByteString -> IO ()) -> Maybe StrictByteString -> IO ()
forall b a. b -> (a -> b) -> Maybe a -> b
maybe IO ()
keepalive StrictByteString -> IO ()
event IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
pump
            -- The connected event says the channel is subscribed.
            IO ()
awaitReady
            StrictByteString -> IO ()
event StrictByteString
"{\"event\":\"connected\",\"message\":\"Stream connected\"}"
            IO ()
pump
  where
    sseHeaders :: ResponseHeaders
sseHeaders =
      [ (HeaderName
hContentType, StrictByteString
"text/event-stream")
      , (HeaderName
hCacheControl, StrictByteString
"no-cache")
      , (HeaderName
"Connection", StrictByteString
"keep-alive")
      , (HeaderName
"X-Accel-Buffering", StrictByteString
"no")
      ]

-- | Idle gap before an SSE client gets a keepalive comment.
sseKeepaliveMicros :: Int
sseKeepaliveMicros :: Int
sseKeepaliveMicros = Int
15_000_000

-- | The event-streaming channel as the hub names it.
eventStreamingChannel :: ByteString
eventStreamingChannel :: StrictByteString
eventStreamingChannel = Text -> StrictByteString
encodeUtf8 Text
Schema.eventStreamingChannel

-- | Cron API handlers.
cronServer
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> CronAPI (AsServerT Handler)
cronServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> CronAPI (AsServerT Handler)
cronServer ArbiterServerConfig m registry
config =
  CronAPI
    { listSchedules :: AsServerT Handler
:- (Summary "List cron schedules"
    :> ("schedules"
        :> (QueryParam "queue" Text :> Get '[JSON] CronSchedulesResponse)))
listSchedules = ArbiterServerConfig m registry
-> Maybe Text -> Handler CronSchedulesResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Maybe Text -> Handler CronSchedulesResponse
listCronSchedulesHandler ArbiterServerConfig m registry
config
    , updateSchedule :: AsServerT Handler
:- (Summary "Override a cron schedule"
    :> (Description
          "A null field clears its override. An absent field keeps it."
        :> (Throws 400 "An override is not valid."
            :> ("schedules"
                :> (Capture "name" Text
                    :> (ReqBody '[JSON] CronScheduleUpdate
                        :> Patch '[JSON] CronScheduleView))))))
updateSchedule = ArbiterServerConfig m registry
-> Text -> CronScheduleUpdate -> Handler CronScheduleView
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text -> CronScheduleUpdate -> Handler CronScheduleView
updateCronScheduleHandler ArbiterServerConfig m registry
config
    , runSchedule :: AsServerT Handler
:- (Summary "Run a cron schedule now"
    :> (Throws
          409
          "The schedule is disabled, or it already has a run that waits to start."
        :> ("schedules"
            :> (Capture "name" Text :> ("run" :> PostNoContent)))))
runSchedule = ArbiterServerConfig m registry -> Text -> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Text -> Handler NoContent
runCronScheduleHandler ArbiterServerConfig m registry
config
    }

-- | List cron schedules, optionally scoped to a queue.
listCronSchedulesHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Maybe Text
  -> Handler CronSchedulesResponse
listCronSchedulesHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Maybe Text -> Handler CronSchedulesResponse
listCronSchedulesHandler ArbiterServerConfig m registry
config Maybe Text
mQueue = do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
  rows <- ArbiterServerConfig m registry
-> m [CronScheduleRow] -> Handler [CronScheduleRow]
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m [CronScheduleRow] -> Handler [CronScheduleRow])
-> m [CronScheduleRow] -> Handler [CronScheduleRow]
forall a b. (a -> b) -> a -> b
$ Text -> Maybe Text -> m [CronScheduleRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> m [CronScheduleRow]
Ops.listCronSchedules Text
schemaName Maybe Text
mQueue
  now <- liftIO getCurrentTime
  pure $ CronSchedulesResponse {schedules = map (cronScheduleView now) rows}

-- | A schedule row with the next tick it fires at. A disabled schedule has none.
cronScheduleView :: UTCTime -> CronScheduleRow -> CronScheduleView
cronScheduleView :: UTCTime -> CronScheduleRow -> CronScheduleView
cronScheduleView UTCTime
now row :: CronScheduleRow
row@CS.CronScheduleRow {enabled :: CronScheduleRow -> Bool
CS.enabled = Bool
isEnabled} =
  CronScheduleView
    { schedule :: CronScheduleRow
schedule = CronScheduleRow
row
    , nextRunAt :: Maybe UTCTime
nextRunAt = do
        Bool -> Maybe ()
forall (f :: * -> *). Alternative f => Bool -> f ()
guard Bool
isEnabled
        Maybe Text -> Text -> UTCTime -> Maybe UTCTime
nextRunFromExpression (CronScheduleRow -> Maybe Text
CS.effectiveTimezone CronScheduleRow
row) (CronScheduleRow -> Text
CS.effectiveExpression CronScheduleRow
row) UTCTime
now
    }

-- | Update a cron schedule.
updateCronScheduleHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Text
  -> CronScheduleUpdate
  -> Handler CronScheduleView
updateCronScheduleHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text -> CronScheduleUpdate -> Handler CronScheduleView
updateCronScheduleHandler ArbiterServerConfig m registry
config Text
name CronScheduleUpdate
update = do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
  result <- ArbiterServerConfig m registry
-> m (Either Text (Maybe CronScheduleRow))
-> Handler (Either Text (Maybe CronScheduleRow))
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m (Either Text (Maybe CronScheduleRow))
 -> Handler (Either Text (Maybe CronScheduleRow)))
-> m (Either Text (Maybe CronScheduleRow))
-> Handler (Either Text (Maybe CronScheduleRow))
forall a b. (a -> b) -> a -> b
$ m (Either Text (Maybe CronScheduleRow))
-> m (Either Text (Maybe CronScheduleRow))
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Either Text (Maybe CronScheduleRow))
 -> m (Either Text (Maybe CronScheduleRow)))
-> m (Either Text (Maybe CronScheduleRow))
-> m (Either Text (Maybe CronScheduleRow))
forall a b. (a -> b) -> a -> b
$ do
    outcome <- Text -> CronScheduleUpdate -> m (Either Text Int64)
forall (m :: * -> *).
MonadArbiter m =>
Text -> CronScheduleUpdate -> m (Either Text Int64)
updateCronScheduleChecked Text
name CronScheduleUpdate
update
    traverse (const (Ops.getCronScheduleByName schemaName name)) outcome

  case result of
    Left Text
err -> ServerError -> Handler CronScheduleView
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err400 {errBody = LBS.fromStrict (encodeUtf8 err)}
    Right Maybe CronScheduleRow
Nothing -> ServerError -> Handler CronScheduleView
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
cronScheduleNotFound
    Right (Just CronScheduleRow
row) -> (UTCTime -> CronScheduleRow -> CronScheduleView)
-> CronScheduleRow -> UTCTime -> CronScheduleView
forall a b c. (a -> b -> c) -> b -> a -> c
flip UTCTime -> CronScheduleRow -> CronScheduleView
cronScheduleView CronScheduleRow
row (UTCTime -> CronScheduleView)
-> Handler UTCTime -> Handler CronScheduleView
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO UTCTime -> Handler UTCTime
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO UTCTime
getCurrentTime

-- | Request an out-of-band run of a cron schedule. A disabled schedule is
-- refused. A schedule with a run already pending is refused.
runCronScheduleHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Text
  -> Handler NoContent
runCronScheduleHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Text -> Handler NoContent
runCronScheduleHandler ArbiterServerConfig m registry
config Text
name = do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
  outcome <- ArbiterServerConfig m registry
-> m RunRequestOutcome -> Handler RunRequestOutcome
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m RunRequestOutcome -> Handler RunRequestOutcome)
-> m RunRequestOutcome -> Handler RunRequestOutcome
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schemaName Text
name
  case outcome of
    RunRequestOutcome
Ops.RunReqNotFound -> ServerError -> Handler NoContent
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
cronScheduleNotFound
    RunRequestOutcome
Ops.RunReqDisabled -> ServerError -> Handler NoContent
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err409 {errBody = "Cron schedule is disabled"}
    RunRequestOutcome
Ops.RunReqPending -> ServerError -> Handler NoContent
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err409 {errBody = "Cron schedule already has a run pending"}
    RunRequestOutcome
Ops.RunReqStamped -> NoContent -> Handler NoContent
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure NoContent
NoContent

-- | Workers API handlers.
workersServer
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> WorkersAPI (AsServerT Handler)
workersServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> WorkersAPI (AsServerT Handler)
workersServer ArbiterServerConfig m registry
config =
  WorkersAPI
    { listWorkers :: AsServerT Handler
:- (Summary "List workers"
    :> (QueryParam "queue" Text
        :> (QueryParam "live" Double :> Get '[JSON] WorkersResponse)))
listWorkers = ArbiterServerConfig m registry
-> Maybe Text -> Maybe Double -> Handler WorkersResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Maybe Text -> Maybe Double -> Handler WorkersResponse
listWorkersHandler ArbiterServerConfig m registry
config
    , pauseWorker :: AsServerT Handler
:- (Summary "Pause a worker"
    :> (Description "The worker does not claim new jobs."
        :> (Capture "id" UUID :> ("pause" :> PostNoContent))))
pauseWorker = ArbiterServerConfig m registry -> Bool -> UUID -> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Bool -> UUID -> Handler NoContent
setWorkerPausedHandler ArbiterServerConfig m registry
config Bool
True
    , resumeWorker :: AsServerT Handler
:- (Summary "Resume a worker"
    :> (Capture "id" UUID :> ("resume" :> PostNoContent)))
resumeWorker = ArbiterServerConfig m registry -> Bool -> UUID -> Handler NoContent
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Bool -> UUID -> Handler NoContent
setWorkerPausedHandler ArbiterServerConfig m registry
config Bool
False
    }

-- | List workers, optionally scoped to a queue and/or to recent heartbeats.
listWorkersHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Maybe Text
  -> Maybe Double
  -> Handler WorkersResponse
listWorkersHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Maybe Text -> Maybe Double -> Handler WorkersResponse
listWorkersHandler ArbiterServerConfig m registry
config Maybe Text
mQueue Maybe Double
mLiveSecs = do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
  rows <- ArbiterServerConfig m registry
-> m [WorkerRow] -> Handler [WorkerRow]
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (m [WorkerRow] -> Handler [WorkerRow])
-> m [WorkerRow] -> Handler [WorkerRow]
forall a b. (a -> b) -> a -> b
$ Text -> Maybe Text -> Maybe NominalDiffTime -> m [WorkerRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> Maybe NominalDiffTime -> m [WorkerRow]
Ops.listWorkers Text
schemaName Maybe Text
mQueue (Double -> NominalDiffTime
forall a b. (Real a, Fractional b) => a -> b
realToFrac (Double -> NominalDiffTime)
-> Maybe Double -> Maybe NominalDiffTime
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Double
mLiveSecs)
  pure $ WorkersResponse {workers = rows}

-- | Set a worker's @paused@ flag. The worker reconciles its local state on
-- the next heartbeat.
setWorkerPausedHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Bool
  -> UUID
  -> Handler NoContent
setWorkerPausedHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Bool -> UUID -> Handler NoContent
setWorkerPausedHandler ArbiterServerConfig m registry
config Bool
pauseFlag UUID
workerId =
  ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
forall (m :: * -> *) (registry :: JobPayloadRegistry).
Monad m =>
ArbiterServerConfig m registry
-> ServerError -> (Text -> m Int64) -> Handler NoContent
mutateOr404 ArbiterServerConfig m registry
config ServerError
workerNotFound ((Text -> m Int64) -> Handler NoContent)
-> (Text -> m Int64) -> Handler NoContent
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
pauseFlag

-- | Rate-limit management/observability handlers.
rateLimitsServer
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> RateLimitsAPI (AsServerT Handler)
rateLimitsServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry -> RateLimitsAPI (AsServerT Handler)
rateLimitsServer ArbiterServerConfig m registry
config =
  RateLimitsAPI
    { listRateLimits :: AsServerT Handler
:- (Summary "List rate-limit policies"
    :> Get '[JSON] RateLimitPoliciesResponse)
listRateLimits = ArbiterServerConfig m registry -> Handler RateLimitPoliciesResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry -> Handler RateLimitPoliciesResponse
listRateLimitsHandler ArbiterServerConfig m registry
config
    , listRateLimitBuckets :: AsServerT Handler
:- (Summary "List the buckets of a rate-limit policy"
    :> (Capture "prefix" Text
        :> ("buckets"
            :> (QueryParam "limit" KeyPageLimit
                :> (QueryParam "offset" Int
                    :> Get '[JSON] RateLimitBucketsResponse)))))
listRateLimitBuckets = ArbiterServerConfig m registry
-> Text
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler RateLimitBucketsResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler RateLimitBucketsResponse
listRateLimitBucketsHandler ArbiterServerConfig m registry
config
    , updateRateLimitPolicy :: AsServerT Handler
:- (Summary "Override a rate-limit policy"
    :> (Description
          "A null field clears its override. An absent field keeps it."
        :> (Throws 400 "An override value is not valid."
            :> (Capture "prefix" Text
                :> (ReqBody '[JSON] RateLimitPolicyUpdate
                    :> Patch '[JSON] RateLimitPolicyView)))))
updateRateLimitPolicy = ArbiterServerConfig m registry
-> Text -> RateLimitPolicyUpdate -> Handler RateLimitPolicyView
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Text -> RateLimitPolicyUpdate -> Handler RateLimitPolicyView
updateRateLimitPolicyHandler ArbiterServerConfig m registry
config
    , resetRateLimitBuckets :: AsServerT Handler
:- (Summary "Refill the buckets of a rate-limit policy"
    :> (Description
          "Refills every bucket under the prefix to full. Returns the number refilled."
        :> (Capture "prefix" Text
            :> ("reset" :> Post '[JSON] RateLimitResetResponse))))
resetRateLimitBuckets = ArbiterServerConfig m registry
-> Text -> Handler RateLimitResetResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Text -> Handler RateLimitResetResponse
resetRateLimitBucketsHandler ArbiterServerConfig m registry
config
    , addRateLimitTokens :: AsServerT Handler
:- (Summary "Add tokens to a rate-limit bucket"
    :> (Description "Wakes the throttled jobs of the key."
        :> (Throws
              400
              "The token count is 0 or less, or the key is not under the prefix."
            :> (Capture "prefix" Text
                :> ("buckets"
                    :> (Capture "key" Text
                        :> ("tokens"
                            :> (ReqBody '[JSON] AddTokensRequest
                                :> Post '[JSON] AddTokensResponse))))))))
addRateLimitTokens = ArbiterServerConfig m registry
-> Text -> Text -> AddTokensRequest -> Handler AddTokensResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Text -> Text -> AddTokensRequest -> Handler AddTokensResponse
addRateLimitTokensHandler ArbiterServerConfig m registry
config
    , pruneRateLimitBuckets :: AsServerT Handler
:- (Summary "Prune idle rate-limit buckets"
    :> (Throws 400 "The idle value is negative."
        :> ("prune"
            :> (QueryParam "idle" Double :> Post '[JSON] PruneResponse))))
pruneRateLimitBuckets = ArbiterServerConfig m registry
-> Maybe Double -> Handler PruneResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Maybe Double -> Handler PruneResponse
pruneRateLimitBucketsHandler ArbiterServerConfig m registry
config
    }

-- | Liveness and readiness handlers.
healthServer
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> HealthAPI (AsServerT Handler)
healthServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> HealthAPI (AsServerT Handler)
healthServer ArbiterServerConfig m registry
config =
  HealthAPI
    { getHealth :: AsServerT Handler
:- (Summary "Check readiness"
    :> (ThrowsBody 503 "The database is not reachable." HealthResponse
        :> Get '[JSON] HealthResponse))
getHealth = ArbiterServerConfig m registry -> Handler HealthResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Handler HealthResponse
healthHandler ArbiterServerConfig m registry
config
    , getLiveness :: AsServerT Handler
:- (Summary "Check liveness"
    :> ("live" :> Get '[JSON] LivenessResponse))
getLiveness = LivenessResponse -> Handler LivenessResponse
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure LivenessResponse {alive :: Bool
alive = Bool
True}
    }

-- | Readiness for probes and the dashboard. An unreachable database is a 503.
-- Both answers carry the same body.
healthHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Handler HealthResponse
healthHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Handler HealthResponse
healthHandler ArbiterServerConfig m registry
config = do
  report <- IO HealthResponse -> Handler HealthResponse
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (ArbiterServerConfig m registry -> IO HealthResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> IO HealthResponse
probeHealth ArbiterServerConfig m registry
config)
  case status report of
    HealthStatus
Ok -> HealthResponse -> Handler HealthResponse
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure HealthResponse
report
    HealthStatus
Down ->
      ServerError -> Handler HealthResponse
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError
        ServerError
err503
          { errBody = encode report
          , errHeaders = [(hContentType, "application/json;charset=utf-8")]
          }

-- | Timed database health probe. Cancellation propagates.
probeHealth
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> IO HealthResponse
probeHealth :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> IO HealthResponse
probeHealth ArbiterServerConfig m registry
config = NominalDiffTime
-> CacheCell HealthResponse
-> IO HealthResponse
-> IO HealthResponse
forall a. NominalDiffTime -> CacheCell a -> IO a -> IO a
cachedFor NominalDiffTime
healthCacheTtl (ArbiterServerConfig m registry -> CacheCell HealthResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> CacheCell HealthResponse
healthCache ArbiterServerConfig m registry
config) (IO HealthResponse -> IO HealthResponse)
-> IO HealthResponse -> IO HealthResponse
forall a b. (a -> b) -> a -> b
$ do
  let schemaName :: Text
schemaName = ArbiterServerConfig m registry -> Text
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> Text
serverSchema ArbiterServerConfig m registry
config
  started <- IO UTCTime
getCurrentTime
  probed <- tryAny (timeout healthProbeMicros (runDb config Health.getPgDbHealth))
  finished <- getCurrentTime
  let elapsedMs = NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (UTCTime -> UTCTime -> NominalDiffTime
diffUTCTime UTCTime
finished UTCTime
started) Double -> Double -> Double
forall a. Num a => a -> a -> a
* Double
millisPerSecond
      reached = Maybe (Maybe PgDbHealth)
-> Either SomeException (Maybe (Maybe PgDbHealth))
-> Maybe (Maybe PgDbHealth)
forall b a. b -> Either a b -> b
fromRight Maybe (Maybe PgDbHealth)
forall a. Maybe a
Nothing Either SomeException (Maybe (Maybe PgDbHealth))
probed
  pure
    HealthResponse
      { status = maybe Down (const Ok) reached
      , schemaName = schemaName
      , checkedAt = finished
      , dbLatencyMs = elapsedMs <$ reached
      , db = join reached
      }

-- | Poll-collapsing TTL for the readiness probe.
healthCacheTtl :: NominalDiffTime
healthCacheTtl :: NominalDiffTime
healthCacheTtl = NominalDiffTime
2

-- | Probe time limit. A pool with no available connection reports @down@ when it
-- lapses. A connect blocked in the driver is not interruptible. The connection
-- string needs its own @connect_timeout@.
healthProbeMicros :: Int
healthProbeMicros :: Int
healthProbeMicros = Int
5_000_000

-- | Milliseconds in one second.
millisPerSecond :: Double
millisPerSecond :: Double
millisPerSecond = Double
1000

-- | Stats are read live by default.
defaultStatsCacheTtl :: NominalDiffTime
defaultStatsCacheTtl :: NominalDiffTime
defaultStatsCacheTtl = NominalDiffTime
0

-- | No minimum gap. Each call runs every ordinary operation. Concurrent callers
-- exclude each other on the gate.
defaultMaintenanceInterval :: NominalDiffTime
defaultMaintenanceInterval :: NominalDiffTime
defaultMaintenanceInterval = NominalDiffTime
0

-- | Keyed TTL cache. 'createArbiterServer' creates each cell.
data CacheCell a = CacheCell
  { forall a. CacheCell a -> TVar (Word, Map Text (UTCTime, a))
cacheEntries :: TVar (Word, Map.Map Text (UTCTime, a))
  , forall a. CacheCell a -> TVar (Set Text)
cacheFilling :: TVar (Set.Set Text)
  }

newCacheCell :: IO (CacheCell a)
newCacheCell :: forall a. IO (CacheCell a)
newCacheCell = TVar (Word, Map Text (UTCTime, a))
-> TVar (Set Text) -> CacheCell a
forall a.
TVar (Word, Map Text (UTCTime, a))
-> TVar (Set Text) -> CacheCell a
CacheCell (TVar (Word, Map Text (UTCTime, a))
 -> TVar (Set Text) -> CacheCell a)
-> IO (TVar (Word, Map Text (UTCTime, a)))
-> IO (TVar (Set Text) -> CacheCell a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (Word, Map Text (UTCTime, a))
-> IO (TVar (Word, Map Text (UTCTime, a)))
forall a. a -> IO (TVar a)
newTVarIO (Word
0, Map Text (UTCTime, a)
forall k a. Map k a
Map.empty) IO (TVar (Set Text) -> CacheCell a)
-> IO (TVar (Set Text)) -> IO (CacheCell a)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Set Text -> IO (TVar (Set Text))
forall a. a -> IO (TVar a)
newTVarIO Set Text
forall a. Set a
Set.empty

-- | Serve the sole entry of a single-key cell.
cachedFor :: NominalDiffTime -> CacheCell a -> IO a -> IO a
cachedFor :: forall a. NominalDiffTime -> CacheCell a -> IO a -> IO a
cachedFor NominalDiffTime
ttl CacheCell a
cell = NominalDiffTime -> CacheCell a -> Text -> IO a -> IO a
forall a. NominalDiffTime -> CacheCell a -> Text -> IO a -> IO a
cachedForKey NominalDiffTime
ttl CacheCell a
cell Text
""

-- | Serve one key. Concurrent misses on a key collapse onto one @produce@. Its
-- write is skipped when 'invalidate' bumped the epoch meanwhile.
cachedForKey :: NominalDiffTime -> CacheCell a -> Text -> IO a -> IO a
cachedForKey :: forall a. NominalDiffTime -> CacheCell a -> Text -> IO a -> IO a
cachedForKey NominalDiffTime
ttl CacheCell a
cell Text
key IO a
produce
  | NominalDiffTime
ttl NominalDiffTime -> NominalDiffTime -> Bool
forall a. Ord a => a -> a -> Bool
<= NominalDiffTime
0 = IO a
produce
  | Bool
otherwise = IO (Maybe a)
fresh IO (Maybe a) -> (Maybe a -> IO a) -> IO a
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IO a -> (a -> IO a) -> Maybe a -> IO a
forall b a. b -> (a -> b) -> Maybe a -> b
maybe IO a
fill a -> IO a
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
  where
    fresh :: IO (Maybe a)
fresh = do
      now <- IO UTCTime
getCurrentTime
      (_, entries) <- readTVarIO (cacheEntries cell)
      pure $ do
        (storedAt, value) <- Map.lookup key entries
        guard (diffUTCTime now storedAt < ttl)
        pure value
    fill :: IO a
fill = IO () -> IO () -> IO a -> IO a
forall a b c. IO a -> IO b -> IO c -> IO c
bracket_ IO ()
acquire IO ()
release (IO (Maybe a)
fresh IO (Maybe a) -> (Maybe a -> IO a) -> IO a
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IO a -> (a -> IO a) -> Maybe a -> IO a
forall b a. b -> (a -> b) -> Maybe a -> b
maybe IO a
store a -> IO a
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure)
    acquire :: IO ()
acquire = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      inflight <- TVar (Set Text) -> STM (Set Text)
forall a. TVar a -> STM a
readTVar (CacheCell a -> TVar (Set Text)
forall a. CacheCell a -> TVar (Set Text)
cacheFilling CacheCell a
cell)
      check (Set.notMember key inflight)
      modifyTVar' (cacheFilling cell) (Set.insert key)
    release :: IO ()
release = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Set Text) -> (Set Text -> Set Text) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (CacheCell a -> TVar (Set Text)
forall a. CacheCell a -> TVar (Set Text)
cacheFilling CacheCell a
cell) (Text -> Set Text -> Set Text
forall a. Ord a => a -> Set a -> Set a
Set.delete Text
key)
    store :: IO a
store = do
      (epoch, _) <- TVar (Word, Map Text (UTCTime, a))
-> IO (Word, Map Text (UTCTime, a))
forall a. TVar a -> IO a
readTVarIO (CacheCell a -> TVar (Word, Map Text (UTCTime, a))
forall a. CacheCell a -> TVar (Word, Map Text (UTCTime, a))
cacheEntries CacheCell a
cell)
      value <- produce
      now <- getCurrentTime
      atomically $ modifyTVar' (cacheEntries cell) $ \(Word
current, Map Text (UTCTime, a)
cached) ->
        if Word
current Word -> Word -> Bool
forall a. Eq a => a -> a -> Bool
== Word
epoch then (Word
current, Text
-> (UTCTime, a) -> Map Text (UTCTime, a) -> Map Text (UTCTime, a)
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert Text
key (UTCTime
now, a
value) Map Text (UTCTime, a)
cached) else (Word
current, Map Text (UTCTime, a)
cached)
      pure value

-- | Increment a cache cell's epoch and clear its entries after an operator mutation.
invalidate :: CacheCell a -> Handler ()
invalidate :: forall a. CacheCell a -> Handler ()
invalidate CacheCell a
cell = IO () -> Handler ()
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> Handler ()) -> IO () -> Handler ()
forall a b. (a -> b) -> a -> b
$ STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Word, Map Text (UTCTime, a))
-> ((Word, Map Text (UTCTime, a)) -> (Word, Map Text (UTCTime, a)))
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (CacheCell a -> TVar (Word, Map Text (UTCTime, a))
forall a. CacheCell a -> TVar (Word, Map Text (UTCTime, a))
cacheEntries CacheCell a
cell) (((Word, Map Text (UTCTime, a)) -> (Word, Map Text (UTCTime, a)))
 -> STM ())
-> ((Word, Map Text (UTCTime, a)) -> (Word, Map Text (UTCTime, a)))
-> STM ()
forall a b. (a -> b) -> a -> b
$ \(Word
epoch, Map Text (UTCTime, a)
_) -> (Word
epoch Word -> Word -> Word
forall a. Num a => a -> a -> a
+ Word
1, Map Text (UTCTime, a)
forall k a. Map k a
Map.empty)

-- | Run an operator mutation, then 'invalidate' the cache it makes stale.
invalidating :: CacheCell a -> Handler b -> Handler b
invalidating :: forall a b. CacheCell a -> Handler b -> Handler b
invalidating CacheCell a
cell Handler b
mutation = Handler b
mutation Handler b -> Handler () -> Handler b
forall a b. Handler a -> Handler b -> Handler a
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f a
<* CacheCell a -> Handler ()
forall a. CacheCell a -> Handler ()
invalidate CacheCell a
cell

-- | Run a read through a stats-TTL cache cell and wrap its result.
cachedRead :: ArbiterServerConfig m registry -> CacheCell a -> (b -> a) -> m b -> Handler a
cachedRead :: forall (m :: * -> *) (registry :: JobPayloadRegistry) a b.
ArbiterServerConfig m registry
-> CacheCell a -> (b -> a) -> m b -> Handler a
cachedRead ArbiterServerConfig m registry
config CacheCell a
cell b -> a
wrap m b
action = IO a -> Handler a
forall a. IO a -> Handler a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO a -> Handler a) -> IO a -> Handler a
forall a b. (a -> b) -> a -> b
$ NominalDiffTime -> CacheCell a -> IO a -> IO a
forall a. NominalDiffTime -> CacheCell a -> IO a -> IO a
cachedFor (ArbiterServerConfig m registry -> NominalDiffTime
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
statsCacheTtl ArbiterServerConfig m registry
config) CacheCell a
cell (b -> a
wrap (b -> a) -> IO b -> IO a
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry -> m b -> IO b
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config m b
action)

-- | List policies with bucket stats and currently-throttled job counts.
listRateLimitsHandler
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> Handler RateLimitPoliciesResponse
listRateLimitsHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry -> Handler RateLimitPoliciesResponse
listRateLimitsHandler ArbiterServerConfig m registry
config =
  ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
-> ([RateLimitPolicyView] -> RateLimitPoliciesResponse)
-> m [RateLimitPolicyView]
-> Handler RateLimitPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry) a b.
ArbiterServerConfig m registry
-> CacheCell a -> (b -> a) -> m b -> Handler a
cachedRead
    ArbiterServerConfig m registry
config
    (ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
rateLimitPoliciesCache ArbiterServerConfig m registry
config)
    (\[RateLimitPolicyView]
views -> RateLimitPoliciesResponse {policies :: [RateLimitPolicyView]
policies = [RateLimitPolicyView]
views})
    m [RateLimitPolicyView]
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m [RateLimitPolicyView]
HL.listRateLimitPolicies

-- | List a prefix's buckets with fill levels, paginated.
listRateLimitBucketsHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Text
  -> Maybe KeyPageLimit
  -> Maybe Int
  -> Handler RateLimitBucketsResponse
listRateLimitBucketsHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler RateLimitBucketsResponse
listRateLimitBucketsHandler ArbiterServerConfig m registry
config Text
prefix = ArbiterServerConfig m registry
-> (Int -> Int -> m [RateLimitBucketView])
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler RateLimitBucketsResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> (Int -> Int -> m [a])
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler (Items a)
keyPage ArbiterServerConfig m registry
config (Text -> Int -> Int -> m [RateLimitBucketView]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Int -> Int -> m [RateLimitBucketView]
HL.listRateLimitBuckets Text
prefix)

-- | Read one page of keys at a limit and offset.
keyPage
  :: ArbiterServerConfig m registry
  -> (Int -> Int -> m [a])
  -> Maybe KeyPageLimit
  -> Maybe Int
  -> Handler (Items a)
keyPage :: forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> (Int -> Int -> m [a])
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler (Items a)
keyPage ArbiterServerConfig m registry
config Int -> Int -> m [a]
page Maybe KeyPageLimit
mLimit Maybe Int
mOffset =
  let (Int
limit, Int
offset) = Maybe KeyPageLimit -> Maybe Int -> (Int, Int)
forall (def :: Nat).
KnownNat def =>
Maybe (PageSize def) -> Maybe Int -> (Int, Int)
validatePagination Maybe KeyPageLimit
mLimit Maybe Int
mOffset
   in (\[a]
rows -> Items {items :: [a]
items = [a]
rows}) ([a] -> Items a) -> Handler [a] -> Handler (Items a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry -> m [a] -> Handler [a]
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (Int -> Int -> m [a]
page Int
limit Int
offset)

-- | Run an action. 'Nothing' throws the given error.
orNotFound
  :: ArbiterServerConfig m registry
  -> m (Maybe a)
  -> ServerError
  -> Handler a
orNotFound :: forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> m (Maybe a) -> ServerError -> Handler a
orNotFound ArbiterServerConfig m registry
config m (Maybe a)
action ServerError
notFound =
  ArbiterServerConfig m registry -> m (Maybe a) -> Handler (Maybe a)
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config m (Maybe a)
action Handler (Maybe a) -> (Maybe a -> Handler a) -> Handler a
forall a b. Handler a -> (a -> Handler b) -> Handler b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Handler a -> (a -> Handler a) -> Maybe a -> Handler a
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (ServerError -> Handler a
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
notFound) a -> Handler a
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure

-- | Set or clear a policy's override params, then return the updated view.
updateRateLimitPolicyHandler
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> Text
  -> RateLimitPolicyUpdate
  -> Handler RateLimitPolicyView
updateRateLimitPolicyHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Text -> RateLimitPolicyUpdate -> Handler RateLimitPolicyView
updateRateLimitPolicyHandler ArbiterServerConfig m registry
config Text
prefix upd :: RateLimitPolicyUpdate
upd@(RateLimitPolicyUpdate Maybe (Maybe Double)
mMax Maybe (Maybe Double)
mRefill Maybe (Maybe Double)
mInterval) = do
  let invalid :: Maybe ByteString
invalid
        | (Double -> Bool) -> Maybe Double -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
< Double
0) (Maybe (Maybe Double) -> Maybe Double
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join Maybe (Maybe Double)
mMax) = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"override max tokens must be >= 0"
        | (Double -> Bool) -> Maybe Double -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
< Double
0) (Maybe (Maybe Double) -> Maybe Double
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join Maybe (Maybe Double)
mRefill) = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"override refill amount must be >= 0"
        | (Double -> Bool) -> Maybe Double -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
<= Double
0) (Maybe (Maybe Double) -> Maybe Double
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join Maybe (Maybe Double)
mInterval) = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"override interval must be > 0"
        | Bool
otherwise = Maybe ByteString
forall a. Maybe a
Nothing
  (ByteString -> Handler (ZonkAny 8))
-> Maybe ByteString -> Handler ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\ByteString
msg -> ServerError -> Handler (ZonkAny 8)
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err400 {errBody = msg}) Maybe ByteString
invalid
  -- An all-absent patch reads the view without rewriting the row.
  let update :: m ()
update = Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ((Maybe (Maybe Double) -> Bool) -> [Maybe (Maybe Double)] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any Maybe (Maybe Double) -> Bool
forall a. Maybe a -> Bool
isJust [Maybe (Maybe Double)
mMax, Maybe (Maybe Double)
mRefill, Maybe (Maybe Double)
mInterval]) (m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Text -> RateLimitPolicyUpdate -> m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> RateLimitPolicyUpdate -> m Int64
HL.updateRateLimitPolicyOverrides Text
prefix RateLimitPolicyUpdate
upd))
  CacheCell RateLimitPoliciesResponse
-> Handler RateLimitPolicyView -> Handler RateLimitPolicyView
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
rateLimitPoliciesCache ArbiterServerConfig m registry
config) (Handler RateLimitPolicyView -> Handler RateLimitPolicyView)
-> Handler RateLimitPolicyView -> Handler RateLimitPolicyView
forall a b. (a -> b) -> a -> b
$
    ArbiterServerConfig m registry
-> m (Maybe RateLimitPolicyView)
-> ServerError
-> Handler RateLimitPolicyView
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> m (Maybe a) -> ServerError -> Handler a
orNotFound ArbiterServerConfig m registry
config (m ()
update m ()
-> m (Maybe RateLimitPolicyView) -> m (Maybe RateLimitPolicyView)
forall a b. m a -> m b -> m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Text -> m (Maybe RateLimitPolicyView)
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> m (Maybe RateLimitPolicyView)
HL.getRateLimitPolicy Text
prefix) ServerError
rateLimitPolicyNotFound

-- | Refill every bucket for a prefix to full. Returns the number reset. 404s an unknown prefix.
resetRateLimitBucketsHandler
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> Text
  -> Handler RateLimitResetResponse
resetRateLimitBucketsHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Text -> Handler RateLimitResetResponse
resetRateLimitBucketsHandler ArbiterServerConfig m registry
config Text
prefix =
  CacheCell RateLimitPoliciesResponse
-> Handler RateLimitResetResponse -> Handler RateLimitResetResponse
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
rateLimitPoliciesCache ArbiterServerConfig m registry
config) (Handler RateLimitResetResponse -> Handler RateLimitResetResponse)
-> Handler RateLimitResetResponse -> Handler RateLimitResetResponse
forall a b. (a -> b) -> a -> b
$
    Int64 -> RateLimitResetResponse
RateLimitResetResponse
      (Int64 -> RateLimitResetResponse)
-> Handler Int64 -> Handler RateLimitResetResponse
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry
-> m (Maybe Int64) -> ServerError -> Handler Int64
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> m (Maybe a) -> ServerError -> Handler a
orNotFound ArbiterServerConfig m registry
config (Text -> m Int64 -> m (Maybe Int64)
forall (m :: * -> *) a.
MonadArbiter m =>
Text -> m a -> m (Maybe a)
onRateLimitPolicy Text
prefix (Text -> m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
Text -> m Int64
HL.resetRateLimitBuckets Text
prefix)) ServerError
rateLimitPolicyNotFound

-- | Run an action for a rate-limit policy. 'Nothing' for an unknown prefix.
onRateLimitPolicy :: (MonadArbiter m) => Text -> m a -> m (Maybe a)
onRateLimitPolicy :: forall (m :: * -> *) a.
MonadArbiter m =>
Text -> m a -> m (Maybe a)
onRateLimitPolicy Text
prefix m a
action =
  Text -> m Bool
forall (m :: * -> *). MonadArbiter m => Text -> m Bool
HL.rateLimitPolicyExists Text
prefix m Bool -> (Bool -> m (Maybe a)) -> m (Maybe a)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Bool
exists -> if Bool
exists then a -> Maybe a
forall a. a -> Maybe a
Just (a -> Maybe a) -> m a -> m (Maybe a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m a
action else Maybe a -> m (Maybe a)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe a
forall a. Maybe a
Nothing

-- | Add tokens to one bucket, named by its full key as the bucket listing shows it,
-- and wake the key's throttled jobs. 404s an unknown prefix. 400s a key outside it.
addRateLimitTokensHandler
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> Text
  -> Text
  -> AddTokensRequest
  -> Handler AddTokensResponse
addRateLimitTokensHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Text -> Text -> AddTokensRequest -> Handler AddTokensResponse
addRateLimitTokensHandler ArbiterServerConfig m registry
config Text
prefix Text
key AddTokensRequest
request = do
  Bool -> Handler () -> Handler ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (AddTokensRequest -> Double
addedTokens AddTokensRequest
request Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
<= Double
0) (Handler () -> Handler ()) -> Handler () -> Handler ()
forall a b. (a -> b) -> a -> b
$ ServerError -> Handler ()
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err400 {errBody = "tokens must be > 0"}
  suffix <-
    Handler Text
-> (Text -> Handler Text) -> Maybe Text -> Handler Text
forall b a. b -> (a -> b) -> Maybe a -> b
maybe
      (ServerError -> Handler Text
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err400 {errBody = "The bucket key does not start with the policy prefix"})
      Text -> Handler Text
forall a. a -> Handler a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
      (Text -> Text -> Maybe Text
T.stripPrefix (Text
prefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
":") Text
key)
  let grant = RateLimitKey -> Double -> m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
RateLimitKey -> Double -> m Int64
HL.addRateLimitTokens RateLimitKey {rlkPrefix :: Text
rlkPrefix = Text
prefix, rlkSuffix :: Text
rlkSuffix = Text
suffix} (AddTokensRequest -> Double
addedTokens AddTokensRequest
request)
  invalidating (rateLimitPoliciesCache config) $
    AddTokensResponse <$> orNotFound config (onRateLimitPolicy prefix grant) rateLimitPolicyNotFound

-- | Delete full buckets idle for @idle@ seconds, the maintenance idle age by default.
pruneRateLimitBucketsHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Maybe Double
  -> Handler PruneResponse
pruneRateLimitBucketsHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Maybe Double -> Handler PruneResponse
pruneRateLimitBucketsHandler ArbiterServerConfig m registry
config Maybe Double
mIdle = do
  let idle :: NominalDiffTime
idle = NominalDiffTime
-> (Double -> NominalDiffTime) -> Maybe Double -> NominalDiffTime
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (ArbiterServerConfig m registry -> NominalDiffTime
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry -> NominalDiffTime
maintenanceBucketIdle ArbiterServerConfig m registry
config) Double -> NominalDiffTime
forall a b. (Real a, Fractional b) => a -> b
realToFrac Maybe Double
mIdle
  Bool -> Handler () -> Handler ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (NominalDiffTime
idle NominalDiffTime -> NominalDiffTime -> Bool
forall a. Ord a => a -> a -> Bool
< NominalDiffTime
0) (Handler () -> Handler ()) -> Handler () -> Handler ()
forall a b. (a -> b) -> a -> b
$ ServerError -> Handler ()
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err400 {errBody = "idle must be >= 0"}
  CacheCell RateLimitPoliciesResponse
-> Handler PruneResponse -> Handler PruneResponse
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell RateLimitPoliciesResponse
rateLimitPoliciesCache ArbiterServerConfig m registry
config) (Handler PruneResponse -> Handler PruneResponse)
-> Handler PruneResponse -> Handler PruneResponse
forall a b. (a -> b) -> a -> b
$
    Int64 -> PruneResponse
PruneResponse (Int64 -> PruneResponse) -> Handler Int64 -> Handler PruneResponse
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry -> m Int64 -> Handler Int64
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config (NominalDiffTime -> m Int64
forall (m :: * -> *). MonadArbiter m => NominalDiffTime -> m Int64
HL.pruneRateLimitBuckets NominalDiffTime
idle)

-- | Concurrency management/observability handlers.
concurrencyServer
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> ConcurrencyAPI (AsServerT Handler)
concurrencyServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> ConcurrencyAPI (AsServerT Handler)
concurrencyServer ArbiterServerConfig m registry
config =
  ConcurrencyAPI
    { listConcurrency :: AsServerT Handler
:- (Summary "List concurrency policies"
    :> Get '[JSON] ConcurrencyPoliciesResponse)
listConcurrency = ArbiterServerConfig m registry
-> Handler ConcurrencyPoliciesResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Handler ConcurrencyPoliciesResponse
listConcurrencyHandler ArbiterServerConfig m registry
config
    , listConcurrencyKeys :: AsServerT Handler
:- (Summary "List the keys of a concurrency policy"
    :> (Capture "prefix" Text
        :> ("keys"
            :> (QueryParam "limit" KeyPageLimit
                :> (QueryParam "offset" Int
                    :> Get '[JSON] ConcurrencyKeysResponse)))))
listConcurrencyKeys = ArbiterServerConfig m registry
-> Text
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler ConcurrencyKeysResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler ConcurrencyKeysResponse
listConcurrencyKeysHandler ArbiterServerConfig m registry
config
    , updateConcurrencyPolicy :: AsServerT Handler
:- (Summary "Override a concurrency limit"
    :> (Description
          "A null overrideLimit clears the override. An absent one keeps it."
        :> (Throws 400 "The override limit is negative."
            :> (Capture "prefix" Text
                :> (ReqBody '[JSON] ConcurrencyPolicyUpdate
                    :> Patch '[JSON] ConcurrencyPolicyView)))))
updateConcurrencyPolicy = ArbiterServerConfig m registry
-> Text -> ConcurrencyPolicyUpdate -> Handler ConcurrencyPolicyView
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text -> ConcurrencyPolicyUpdate -> Handler ConcurrencyPolicyView
updateConcurrencyPolicyHandler ArbiterServerConfig m registry
config
    , reconcileConcurrency :: AsServerT Handler
:- (Summary "Recount in-flight jobs per concurrency key"
    :> (Description
          "Repairs each in-flight count from the live jobs. Returns the rows repaired."
        :> ("reconcile" :> Post '[JSON] ConcurrencyReconcileResponse)))
reconcileConcurrency = ArbiterServerConfig m registry
-> Handler ConcurrencyReconcileResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Handler ConcurrencyReconcileResponse
reconcileConcurrencyHandler ArbiterServerConfig m registry
config
    , pruneConcurrencyKeys :: AsServerT Handler
:- (Summary "Prune drained concurrency keys"
    :> (Description "Deletes keys with no live job."
        :> ("prune" :> Post '[JSON] PruneResponse)))
pruneConcurrencyKeys = ArbiterServerConfig m registry -> Handler PruneResponse
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry -> Handler PruneResponse
pruneConcurrencyKeysHandler ArbiterServerConfig m registry
config
    }

-- | List concurrency policies with their default/override limit and live key/in-flight stats.
listConcurrencyHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Handler ConcurrencyPoliciesResponse
listConcurrencyHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Handler ConcurrencyPoliciesResponse
listConcurrencyHandler ArbiterServerConfig m registry
config =
  ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
-> ([ConcurrencyPolicyView] -> ConcurrencyPoliciesResponse)
-> m [ConcurrencyPolicyView]
-> Handler ConcurrencyPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry) a b.
ArbiterServerConfig m registry
-> CacheCell a -> (b -> a) -> m b -> Handler a
cachedRead
    ArbiterServerConfig m registry
config
    (ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
concurrencyPoliciesCache ArbiterServerConfig m registry
config)
    (\[ConcurrencyPolicyView]
views -> ConcurrencyPoliciesResponse {policies :: [ConcurrencyPolicyView]
policies = [ConcurrencyPolicyView]
views})
    m [ConcurrencyPolicyView]
forall (m :: * -> *). MonadArbiter m => m [ConcurrencyPolicyView]
HL.listConcurrencyPolicies

-- | List a prefix's keys with in-flight fill levels, paginated.
listConcurrencyKeysHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Text
  -> Maybe KeyPageLimit
  -> Maybe Int
  -> Handler ConcurrencyKeysResponse
listConcurrencyKeysHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler ConcurrencyKeysResponse
listConcurrencyKeysHandler ArbiterServerConfig m registry
config Text
prefix = ArbiterServerConfig m registry
-> (Int -> Int -> m [ConcurrencyKeyView])
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler ConcurrencyKeysResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> (Int -> Int -> m [a])
-> Maybe KeyPageLimit
-> Maybe Int
-> Handler (Items a)
keyPage ArbiterServerConfig m registry
config (Text -> Int -> Int -> m [ConcurrencyKeyView]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Int -> Int -> m [ConcurrencyKeyView]
HL.listConcurrencyKeys Text
prefix)

-- | Set or clear a concurrency policy's override limit, then return the updated view.
updateConcurrencyPolicyHandler
  :: forall registry m
   . (HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> Text
  -> ConcurrencyPolicyUpdate
  -> Handler ConcurrencyPolicyView
updateConcurrencyPolicyHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> Text -> ConcurrencyPolicyUpdate -> Handler ConcurrencyPolicyView
updateConcurrencyPolicyHandler ArbiterServerConfig m registry
config Text
prefix upd :: ConcurrencyPolicyUpdate
upd@(ConcurrencyPolicyUpdate Maybe (Maybe Int32)
mLimit) = do
  Bool -> Handler () -> Handler ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ((Int32 -> Bool) -> Maybe Int32 -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Int32 -> Int32 -> Bool
forall a. Ord a => a -> a -> Bool
< Int32
0) (Maybe (Maybe Int32) -> Maybe Int32
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join Maybe (Maybe Int32)
mLimit)) (Handler () -> Handler ()) -> Handler () -> Handler ()
forall a b. (a -> b) -> a -> b
$ ServerError -> Handler ()
forall a. ServerError -> Handler a
forall e (m :: * -> *) a. MonadError e m => e -> m a
throwError ServerError
err400 {errBody = "override limit must be >= 0"}
  -- An absent overrideLimit reads the view without rewriting the row.
  let action :: m (Maybe ConcurrencyPolicyView)
action = Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Maybe (Maybe Int32) -> Bool
forall a. Maybe a -> Bool
isJust Maybe (Maybe Int32)
mLimit) (m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Text -> ConcurrencyPolicyUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> ConcurrencyPolicyUpdate -> m Int64
HL.updateConcurrencyPolicyOverrides Text
prefix ConcurrencyPolicyUpdate
upd)) m ()
-> m (Maybe ConcurrencyPolicyView)
-> m (Maybe ConcurrencyPolicyView)
forall a b. m a -> m b -> m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Text -> m (Maybe ConcurrencyPolicyView)
forall (m :: * -> *).
MonadArbiter m =>
Text -> m (Maybe ConcurrencyPolicyView)
HL.getConcurrencyPolicy Text
prefix
  CacheCell ConcurrencyPoliciesResponse
-> Handler ConcurrencyPolicyView -> Handler ConcurrencyPolicyView
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
concurrencyPoliciesCache ArbiterServerConfig m registry
config) (Handler ConcurrencyPolicyView -> Handler ConcurrencyPolicyView)
-> Handler ConcurrencyPolicyView -> Handler ConcurrencyPolicyView
forall a b. (a -> b) -> a -> b
$
    ArbiterServerConfig m registry
-> m (Maybe ConcurrencyPolicyView)
-> ServerError
-> Handler ConcurrencyPolicyView
forall (m :: * -> *) (registry :: JobPayloadRegistry) a.
ArbiterServerConfig m registry
-> m (Maybe a) -> ServerError -> Handler a
orNotFound ArbiterServerConfig m registry
config m (Maybe ConcurrencyPolicyView)
action ServerError
concurrencyPolicyNotFound

-- | Recompute every key's in-flight count from live jobs. Returns rows repaired.
reconcileConcurrencyHandler
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> Handler ConcurrencyReconcileResponse
reconcileConcurrencyHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> Handler ConcurrencyReconcileResponse
reconcileConcurrencyHandler ArbiterServerConfig m registry
config =
  CacheCell ConcurrencyPoliciesResponse
-> Handler ConcurrencyReconcileResponse
-> Handler ConcurrencyReconcileResponse
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
concurrencyPoliciesCache ArbiterServerConfig m registry
config) (Handler ConcurrencyReconcileResponse
 -> Handler ConcurrencyReconcileResponse)
-> Handler ConcurrencyReconcileResponse
-> Handler ConcurrencyReconcileResponse
forall a b. (a -> b) -> a -> b
$
    Int64 -> ConcurrencyReconcileResponse
ConcurrencyReconcileResponse (Int64 -> ConcurrencyReconcileResponse)
-> Handler Int64 -> Handler ConcurrencyReconcileResponse
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry -> m Int64 -> Handler Int64
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
HL.reconcileConcurrencyCounts

-- | Delete drained concurrency keys with no live job. Returns rows deleted.
pruneConcurrencyKeysHandler
  :: forall registry m
   . (HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> Handler PruneResponse
pruneConcurrencyKeysHandler :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry -> Handler PruneResponse
pruneConcurrencyKeysHandler ArbiterServerConfig m registry
config =
  CacheCell ConcurrencyPoliciesResponse
-> Handler PruneResponse -> Handler PruneResponse
forall a b. CacheCell a -> Handler b -> Handler b
invalidating (ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
forall (m :: * -> *) (registry :: JobPayloadRegistry).
ArbiterServerConfig m registry
-> CacheCell ConcurrencyPoliciesResponse
concurrencyPoliciesCache ArbiterServerConfig m registry
config) (Handler PruneResponse -> Handler PruneResponse)
-> Handler PruneResponse -> Handler PruneResponse
forall a b. (a -> b) -> a -> b
$
    Int64 -> PruneResponse
PruneResponse (Int64 -> PruneResponse) -> Handler Int64 -> Handler PruneResponse
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ArbiterServerConfig m registry -> m Int64 -> Handler Int64
forall (n :: * -> *) (m :: * -> *) (registry :: JobPayloadRegistry)
       a.
MonadIO n =>
ArbiterServerConfig m registry -> m a -> n a
runDb ArbiterServerConfig m registry
config m Int64
forall (m :: * -> *).
(MonadArbiter m, RegistryTables (RegistryOf m)) =>
m Int64
HL.pruneConcurrencyKeys

-- | Server for the shared top-level routes.
sharedServer
  :: forall registry m
   . (HL.RegistryAdmissionPolicies registry, HasRegistry m registry, RegistryTables registry)
  => ArbiterServerConfig m registry
  -> ServerT SharedAPI Handler
sharedServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(RegistryAdmissionPolicies registry, HasRegistry m registry,
 RegistryTables registry) =>
ArbiterServerConfig m registry -> ServerT SharedAPI Handler
sharedServer ArbiterServerConfig m registry
config =
  forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
Proxy registry
-> ArbiterServerConfig m registry -> QueuesAPI (AsServerT Handler)
queuesServer @registry (forall (t :: JobPayloadRegistry). Proxy t
forall {k} (t :: k). Proxy t
Proxy @registry) ArbiterServerConfig m registry
config
    QueuesAPI (AsServerT Handler)
-> (MaintenanceAPI (AsServerT Handler)
    :<|> (Tagged Handler Application
          :<|> (CronAPI (AsServerT Handler)
                :<|> (WorkersAPI (AsServerT Handler)
                      :<|> (RateLimitsAPI (AsServerT Handler)
                            :<|> (ConcurrencyAPI (AsServerT Handler)
                                  :<|> HealthAPI (AsServerT Handler)))))))
-> QueuesAPI (AsServerT Handler)
   :<|> (MaintenanceAPI (AsServerT Handler)
         :<|> (Tagged Handler Application
               :<|> (CronAPI (AsServerT Handler)
                     :<|> (WorkersAPI (AsServerT Handler)
                           :<|> (RateLimitsAPI (AsServerT Handler)
                                 :<|> (ConcurrencyAPI (AsServerT Handler)
                                       :<|> HealthAPI (AsServerT Handler)))))))
forall a b. a -> b -> a :<|> b
:<|> forall (registry :: JobPayloadRegistry) (m :: * -> *).
(RegistryAdmissionPolicies registry, HasRegistry m registry,
 RegistryTables registry) =>
ArbiterServerConfig m registry
-> MaintenanceAPI (AsServerT Handler)
maintenanceServer @registry ArbiterServerConfig m registry
config
    MaintenanceAPI (AsServerT Handler)
-> (Tagged Handler Application
    :<|> (CronAPI (AsServerT Handler)
          :<|> (WorkersAPI (AsServerT Handler)
                :<|> (RateLimitsAPI (AsServerT Handler)
                      :<|> (ConcurrencyAPI (AsServerT Handler)
                            :<|> HealthAPI (AsServerT Handler))))))
-> MaintenanceAPI (AsServerT Handler)
   :<|> (Tagged Handler Application
         :<|> (CronAPI (AsServerT Handler)
               :<|> (WorkersAPI (AsServerT Handler)
                     :<|> (RateLimitsAPI (AsServerT Handler)
                           :<|> (ConcurrencyAPI (AsServerT Handler)
                                 :<|> HealthAPI (AsServerT Handler))))))
forall a b. a -> b -> a :<|> b
:<|> ArbiterServerConfig m registry -> Tagged Handler Application
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> Tagged Handler Application
eventsServer ArbiterServerConfig m registry
config
    Tagged Handler Application
-> (CronAPI (AsServerT Handler)
    :<|> (WorkersAPI (AsServerT Handler)
          :<|> (RateLimitsAPI (AsServerT Handler)
                :<|> (ConcurrencyAPI (AsServerT Handler)
                      :<|> HealthAPI (AsServerT Handler)))))
-> Tagged Handler Application
   :<|> (CronAPI (AsServerT Handler)
         :<|> (WorkersAPI (AsServerT Handler)
               :<|> (RateLimitsAPI (AsServerT Handler)
                     :<|> (ConcurrencyAPI (AsServerT Handler)
                           :<|> HealthAPI (AsServerT Handler)))))
forall a b. a -> b -> a :<|> b
:<|> ArbiterServerConfig m registry -> CronAPI (AsServerT Handler)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> CronAPI (AsServerT Handler)
cronServer ArbiterServerConfig m registry
config
    CronAPI (AsServerT Handler)
-> (WorkersAPI (AsServerT Handler)
    :<|> (RateLimitsAPI (AsServerT Handler)
          :<|> (ConcurrencyAPI (AsServerT Handler)
                :<|> HealthAPI (AsServerT Handler))))
-> CronAPI (AsServerT Handler)
   :<|> (WorkersAPI (AsServerT Handler)
         :<|> (RateLimitsAPI (AsServerT Handler)
               :<|> (ConcurrencyAPI (AsServerT Handler)
                     :<|> HealthAPI (AsServerT Handler))))
forall a b. a -> b -> a :<|> b
:<|> ArbiterServerConfig m registry -> WorkersAPI (AsServerT Handler)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> WorkersAPI (AsServerT Handler)
workersServer ArbiterServerConfig m registry
config
    WorkersAPI (AsServerT Handler)
-> (RateLimitsAPI (AsServerT Handler)
    :<|> (ConcurrencyAPI (AsServerT Handler)
          :<|> HealthAPI (AsServerT Handler)))
-> WorkersAPI (AsServerT Handler)
   :<|> (RateLimitsAPI (AsServerT Handler)
         :<|> (ConcurrencyAPI (AsServerT Handler)
               :<|> HealthAPI (AsServerT Handler)))
forall a b. a -> b -> a :<|> b
:<|> ArbiterServerConfig m registry -> RateLimitsAPI (AsServerT Handler)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry -> RateLimitsAPI (AsServerT Handler)
rateLimitsServer ArbiterServerConfig m registry
config
    RateLimitsAPI (AsServerT Handler)
-> (ConcurrencyAPI (AsServerT Handler)
    :<|> HealthAPI (AsServerT Handler))
-> RateLimitsAPI (AsServerT Handler)
   :<|> (ConcurrencyAPI (AsServerT Handler)
         :<|> HealthAPI (AsServerT Handler))
forall a b. a -> b -> a :<|> b
:<|> ArbiterServerConfig m registry
-> ConcurrencyAPI (AsServerT Handler)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(HasRegistry m registry, RegistryTables registry) =>
ArbiterServerConfig m registry
-> ConcurrencyAPI (AsServerT Handler)
concurrencyServer ArbiterServerConfig m registry
config
    ConcurrencyAPI (AsServerT Handler)
-> HealthAPI (AsServerT Handler)
-> ConcurrencyAPI (AsServerT Handler)
   :<|> HealthAPI (AsServerT Handler)
forall a b. a -> b -> a :<|> b
:<|> ArbiterServerConfig m registry -> HealthAPI (AsServerT Handler)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry -> HealthAPI (AsServerT Handler)
healthServer ArbiterServerConfig m registry
config

-- | Builds a registry's per-queue server implementations. @registry@ is the whole
-- registry. @reg@ is the part still to build. Each result type needs @FromJSON@ and an
-- 'EncodeJobResult' instance. Each payload type needs 'JobPayload'.
class BuildServer registry (reg :: JobPayloadRegistry) where
  -- | The server for the queues in @reg@, then the shared routes.
  buildServer :: (HasRegistry m registry) => ArbiterServerConfig m registry -> ServerT (RegistryToAPI reg) Handler

-- The empty registry builds the shared top-level routes alone.
instance
  (HL.RegistryAdmissionPolicies registry, RegistryTables registry)
  => BuildServer registry '[]
  where
  buildServer :: forall (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> ServerT (RegistryToAPI '[]) Handler
buildServer = ArbiterServerConfig m registry -> ServerT SharedAPI Handler
ArbiterServerConfig m registry
-> ServerT (RegistryToAPI '[]) Handler
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(RegistryAdmissionPolicies registry, HasRegistry m registry,
 RegistryTables registry) =>
ArbiterServerConfig m registry -> ServerT SharedAPI Handler
sharedServer

-- One table builds its endpoints, then the rest of the registry.
instance
  ( BuildServer registry rest
  , EncodeJobResult (SpecResult spec)
  , JobPayload (SpecPayload spec)
  , KnownSymbol (SpecName spec)
  )
  => BuildServer registry (spec ': rest)
  where
  buildServer :: forall (m :: * -> *).
HasRegistry m registry =>
ArbiterServerConfig m registry
-> ServerT (RegistryToAPI (spec : rest)) Handler
buildServer ArbiterServerConfig m registry
config =
    let tableName :: Text
tableName = String -> Text
T.pack (String -> Text) -> String -> Text
forall a b. (a -> b) -> a -> b
$ Proxy (SpecName spec) -> String
forall (n :: Symbol) (proxy :: Symbol -> *).
KnownSymbol n =>
proxy n -> String
symbolVal (forall {k} (t :: k). Proxy t
forall (t :: Symbol). Proxy t
Proxy @(SpecName spec))
     in forall (registry :: JobPayloadRegistry) payload result
       (m :: * -> *).
(EncodeJobResult result, HasRegistry m registry,
 JobPayload payload) =>
Text
-> ArbiterServerConfig m registry
-> TableAPI payload result (AsServerT Handler)
tableServer @registry @(SpecPayload spec) @(SpecResult spec) Text
tableName ArbiterServerConfig m registry
config
          TableAPI (SpecPayload spec) (SpecResult spec) (AsServerT Handler)
-> ServerT (RegistryToAPI rest) Handler
-> TableAPI
     (SpecPayload spec) (SpecResult spec) (AsServerT Handler)
   :<|> ServerT (RegistryToAPI rest) Handler
forall a b. a -> b -> a :<|> b
:<|> forall (registry :: JobPayloadRegistry) (reg :: JobPayloadRegistry)
       (m :: * -> *).
(BuildServer registry reg, HasRegistry m registry) =>
ArbiterServerConfig m registry
-> ServerT (RegistryToAPI reg) Handler
buildServer @registry @rest ArbiterServerConfig m registry
config

-- | Complete Arbiter server at @\/api\/...@.
arbiterServer
  :: forall registry m
   . (BuildServer registry registry, HasRegistry m registry)
  => ArbiterServerConfig m registry
  -> ServerT (ArbiterAPI registry) Handler
arbiterServer :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(BuildServer registry registry, HasRegistry m registry) =>
ArbiterServerConfig m registry
-> ServerT (ArbiterAPI registry) Handler
arbiterServer = forall (registry :: JobPayloadRegistry) (reg :: JobPayloadRegistry)
       (m :: * -> *).
(BuildServer registry reg, HasRegistry m registry) =>
ArbiterServerConfig m registry
-> ServerT (RegistryToAPI reg) Handler
buildServer @registry @registry

-- | Hoisted server for integration into a route tree using a custom monad.
arbiterServerHoisted
  :: forall registry m n
   . ( BuildServer registry registry
     , HasRegistry m registry
     , HasServer (ArbiterAPI registry) '[]
     )
  => (forall x. Handler x -> n x)
  -> ArbiterServerConfig m registry
  -> ServerT (ArbiterAPI registry) n
arbiterServerHoisted :: forall (registry :: JobPayloadRegistry) (m :: * -> *)
       (n :: * -> *).
(BuildServer registry registry, HasRegistry m registry,
 HasServer (ArbiterAPI registry) '[]) =>
(forall x. Handler x -> n x)
-> ArbiterServerConfig m registry
-> ServerT (ArbiterAPI registry) n
arbiterServerHoisted forall x. Handler x -> n x
natTrans ArbiterServerConfig m registry
config =
  Proxy (ArbiterAPI registry)
-> (forall x. Handler x -> n x)
-> ServerT (ArbiterAPI registry) Handler
-> ServerT (ArbiterAPI registry) n
forall {k} (api :: k) (m :: * -> *) (n :: * -> *).
HasServer api '[] =>
Proxy api
-> (forall x. m x -> n x) -> ServerT api m -> ServerT api n
hoistServer (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @(ArbiterAPI registry)) Handler x -> n x
forall x. Handler x -> n x
natTrans (ArbiterServerConfig m registry
-> ServerT (ArbiterAPI registry) Handler
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(BuildServer registry registry, HasRegistry m registry) =>
ArbiterServerConfig m registry
-> ServerT (ArbiterAPI registry) Handler
arbiterServer ArbiterServerConfig m registry
config)

-- | Convert to WAI Application.
arbiterApp
  :: forall registry m
   . ( BuildServer registry registry
     , HasRegistry m registry
     , HasServer (ArbiterAPI registry) '[]
     )
  => ArbiterServerConfig m registry
  -> Application
arbiterApp :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(BuildServer registry registry, HasRegistry m registry,
 HasServer (ArbiterAPI registry) '[]) =>
ArbiterServerConfig m registry -> Application
arbiterApp ArbiterServerConfig m registry
config =
  Proxy (ArbiterAPI registry)
-> Server (ArbiterAPI registry) -> Application
forall {k} (api :: k).
HasServer api '[] =>
Proxy api -> Server api -> Application
serve (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @(ArbiterAPI registry)) (ArbiterServerConfig m registry -> Server (ArbiterAPI registry)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(BuildServer registry registry, HasRegistry m registry) =>
ArbiterServerConfig m registry
-> ServerT (ArbiterAPI registry) Handler
arbiterServer ArbiterServerConfig m registry
config)

-- | Run the API server on a port. Prints a start line to stdout.
runArbiterAPI
  :: forall registry m
   . ( BuildServer registry registry
     , HasRegistry m registry
     , HasServer (ArbiterAPI registry) '[]
     )
  => Port
  -> ArbiterServerConfig m registry
  -> IO ()
runArbiterAPI :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
(BuildServer registry registry, HasRegistry m registry,
 HasServer (ArbiterAPI registry) '[]) =>
Int -> ArbiterServerConfig m registry -> IO ()
runArbiterAPI Int
port ArbiterServerConfig m registry
config = do
  String -> IO ()
putStrLn (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"Starting Arbiter API server on port " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Int -> String
forall a. Show a => a -> String
show Int
port
  let settings :: Settings
settings = Int -> Settings -> Settings
setPort Int
port Settings
defaultSettings
  Settings -> Application -> IO ()
runSettings Settings
settings (ArbiterServerConfig m registry -> Application
forall (registry :: JobPayloadRegistry) (m :: * -> *).
(BuildServer registry registry, HasRegistry m registry,
 HasServer (ArbiterAPI registry) '[]) =>
ArbiterServerConfig m registry -> Application
arbiterApp ArbiterServerConfig m registry
config)

-- | The filters every listing takes: parent, group, kind and payload search.
listingFilters :: Maybe Int64 -> Maybe Text -> Maybe Text -> Maybe Text -> [Maybe JobFilter]
listingFilters :: Maybe Int64
-> Maybe Text -> Maybe Text -> Maybe Text -> [Maybe JobFilter]
listingFilters Maybe Int64
mParentId Maybe Text
mGroupKey Maybe Text
mKind Maybe Text
mPayload =
  [ Int64 -> JobFilter
FilterParentId (Int64 -> JobFilter) -> Maybe Int64 -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Int64
mParentId
  , Text -> JobFilter
FilterGroupKey (Text -> JobFilter) -> Maybe Text -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text
mGroupKey
  , Text -> JobFilter
FilterKind (Text -> JobFilter) -> Maybe Text -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text -> Maybe Text
nonBlank Maybe Text
mKind
  , Text -> JobFilter
FilterPayloadText (Text -> JobFilter) -> Maybe Text -> Maybe JobFilter
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Text -> Maybe Text
nonBlank Maybe Text
mPayload
  ]

-- | Remove an empty search parameter.
nonBlank :: Maybe Text -> Maybe Text
nonBlank :: Maybe Text -> Maybe Text
nonBlank = (Text -> Bool) -> Maybe Text -> Maybe Text
forall (m :: * -> *) a. MonadPlus m => (a -> Bool) -> m a -> m a
mfilter (Bool -> Bool
not (Bool -> Bool) -> (Text -> Bool) -> Text -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Bool
T.null (Text -> Bool) -> (Text -> Text) -> Text -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Text
T.strip)

-- | Clamp the limit to 'pageLimitRange' with its default, and keep the offset non-negative.
validatePagination :: forall def. (KnownNat def) => Maybe (PageSize def) -> Maybe Int -> (Int, Int)
validatePagination :: forall (def :: Nat).
KnownNat def =>
Maybe (PageSize def) -> Maybe Int -> (Int, Int)
validatePagination Maybe (PageSize def)
mLimit Maybe Int
mOffset =
  ((Int, Int) -> Int -> Int
forall a. Ord a => (a, a) -> a -> a
clamp (Int, Int)
pageLimitRange (Int -> (PageSize def -> Int) -> Maybe (PageSize def) -> Int
forall b a. b -> (a -> b) -> Maybe a -> b
maybe (Proxy def -> Int
forall (def :: Nat) (proxy :: Nat -> *).
KnownNat def =>
proxy def -> Int
pageSizeDefault (forall (t :: Nat). Proxy t
forall {k} (t :: k). Proxy t
Proxy @def)) PageSize def -> Int
forall (def :: Nat). PageSize def -> Int
unPageSize Maybe (PageSize def)
mLimit), Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
0 (Int -> Maybe Int -> Int
forall a. a -> Maybe a -> a
fromMaybe Int
0 Maybe Int
mOffset))