{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE QuasiQuotes #-}
{-# OPTIONS_HADDOCK not-home #-}

-- | Internal to the arbiter packages. Not covered by the PVP.
--
-- Jobs SQL templates.
module Arbiter.Core.Sql.Jobs
  ( JobFilter (..)
  , FilterTable (..)
  , JobSortColumn (..)
  , jobSortColumnName
  , DLQSortColumn (..)
  , dlqSortColumnName
  , ArchiveSortColumn (..)
  , archiveSortColumnName
  , SortDir (..)
  , sortDirName
  , throttledPredicateSQL
  , jobStatusCaseSQL
  , claimablePred
  , listJobsFilteredSQL
  , listJobsWithStatusSQL
  , countJobsFilteredSQL
  , getJobByIdWithStatusSQL
  , buildJobsOrderBy
  , buildDLQOrderBy
  , buildArchiveOrderBy
  , countWhereSQL
  , buildWhereClause
  , jobColsExceptId
  , dlqCarriedCols
  , requeuedCols
  , requeuedValsEditing
  , enqueuedAgainCols
  , enqueuedAgainValsEditing
  , jobReadColumns
  , insertJobSQL
  , insertJobReplaceSQL
  , insertJobsBatchSQL
  , insertJobsBatchSQL_
  , getJobByIdSQL
  , getJobByDedupKeySQL
  , cancelJobSQL
  , unionAllOverQueueTables
  ) where

import Data.Int (Int64)
import Data.Maybe (fromMaybe)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (UTCTime)
import Data.UUID.Types (UUID)
import NeatInterpolation (text)

import Arbiter.Core.Admission (excludedAssignment)
import Arbiter.Core.Codec (codecColumns, jobRowCodec, joinColumns, writeColumnNames)
import Arbiter.Core.Job.Schema
  ( SchemaName
  , TableName
  , jobQueueDLQTable
  , jobQueueTable
  )
import Arbiter.Core.Job.Types (JobRead, JobStatus, Stored, jobStatusToText)
import Arbiter.Core.Sql.Groups (settleGroupLocksCte)
import Arbiter.Core.Sql.QQ (sql)
import Arbiter.Core.Sql.Query (Query, mwhen, raw, rows, sepBy)
import Arbiter.Core.Sql.Tree (wakeParentCte)
import Arbiter.Core.SqlLiterals (attemptsLeftSQL, attemptsSpentSQL, defaultMaxAttemptsSQL, minMaxAttemptsSQL)

-- | A narrowing predicate on a job listing. Filters combine with AND.
data JobFilter
  = -- | Jobs in this group.
    FilterGroupKey Text
  | -- | Children of this parent.
    FilterParentId Int64
  | -- | Jobs with no parent.
    FilterRootsOnly
  | -- | Jobs with this derived status. Matches nothing on the DLQ or archive.
    FilterStatus JobStatus
  | -- | The row with this primary key.
    FilterId Int64
  | -- | The row with this original job id. On the main table, the row with this id.
    FilterJobId Int64
  | -- | Jobs claimed by this worker.
    FilterClaimedBy UUID
  | -- | Jobs with this kind label.
    FilterKind Text
  | -- | Jobs whose payload text holds this substring, ignoring case.
    FilterPayloadText Text
  | -- | Jobs whose last error holds this substring, ignoring case.
    FilterErrorText Text
  | -- | Jobs under this rate-limit prefix.
    FilterRateLimitPrefix Text
  | -- | Jobs under this concurrency prefix.
    FilterConcurrencyPrefix Text
  | -- | Jobs inserted at or after this time.
    FilterInsertedAfter UTCTime
  | -- | Jobs inserted before this time.
    FilterInsertedBefore UTCTime
  | -- | Jobs completed at or after this time. Matches nothing outside the archive.
    FilterCompletedAfter UTCTime
  | -- | Jobs completed before this time. Matches nothing outside the archive.
    FilterCompletedBefore UTCTime
  deriving stock (JobFilter -> JobFilter -> Bool
(JobFilter -> JobFilter -> Bool)
-> (JobFilter -> JobFilter -> Bool) -> Eq JobFilter
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: JobFilter -> JobFilter -> Bool
== :: JobFilter -> JobFilter -> Bool
$c/= :: JobFilter -> JobFilter -> Bool
/= :: JobFilter -> JobFilter -> Bool
Eq, Int -> JobFilter -> ShowS
[JobFilter] -> ShowS
JobFilter -> String
(Int -> JobFilter -> ShowS)
-> (JobFilter -> String)
-> ([JobFilter] -> ShowS)
-> Show JobFilter
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> JobFilter -> ShowS
showsPrec :: Int -> JobFilter -> ShowS
$cshow :: JobFilter -> String
show :: JobFilter -> String
$cshowList :: [JobFilter] -> ShowS
showList :: [JobFilter] -> ShowS
Show)

-- | The table a 'JobFilter' narrows.
data FilterTable
  = MainTable
  | DLQTable
  | ArchiveTable
  deriving stock (FilterTable -> FilterTable -> Bool
(FilterTable -> FilterTable -> Bool)
-> (FilterTable -> FilterTable -> Bool) -> Eq FilterTable
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: FilterTable -> FilterTable -> Bool
== :: FilterTable -> FilterTable -> Bool
$c/= :: FilterTable -> FilterTable -> Bool
/= :: FilterTable -> FilterTable -> Bool
Eq, Int -> FilterTable -> ShowS
[FilterTable] -> ShowS
FilterTable -> String
(Int -> FilterTable -> ShowS)
-> (FilterTable -> String)
-> ([FilterTable] -> ShowS)
-> Show FilterTable
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> FilterTable -> ShowS
showsPrec :: Int -> FilterTable -> ShowS
$cshow :: FilterTable -> String
show :: FilterTable -> String
$cshowList :: [FilterTable] -> ShowS
showList :: [FilterTable] -> ShowS
Show)

-- | Sortable columns on the main jobs table.
data JobSortColumn
  = -- | By @id@.
    JsId
  | -- | By @priority@.
    JsPriority
  | -- | By @attempts@.
    JsAttempts
  | -- | By @inserted_at@.
    JsInsertedAt
  | -- | By @not_visible_until@.
    JsNotVisibleUntil
  | -- | By @group_key@.
    JsGroupKey
  | -- | By @parent_id@.
    JsParentId
  | -- | By @last_attempted_at@.
    JsLastAttemptedAt
  deriving stock (JobSortColumn
JobSortColumn -> JobSortColumn -> Bounded JobSortColumn
forall a. a -> a -> Bounded a
$cminBound :: JobSortColumn
minBound :: JobSortColumn
$cmaxBound :: JobSortColumn
maxBound :: JobSortColumn
Bounded, Int -> JobSortColumn
JobSortColumn -> Int
JobSortColumn -> [JobSortColumn]
JobSortColumn -> JobSortColumn
JobSortColumn -> JobSortColumn -> [JobSortColumn]
JobSortColumn -> JobSortColumn -> JobSortColumn -> [JobSortColumn]
(JobSortColumn -> JobSortColumn)
-> (JobSortColumn -> JobSortColumn)
-> (Int -> JobSortColumn)
-> (JobSortColumn -> Int)
-> (JobSortColumn -> [JobSortColumn])
-> (JobSortColumn -> JobSortColumn -> [JobSortColumn])
-> (JobSortColumn -> JobSortColumn -> [JobSortColumn])
-> (JobSortColumn
    -> JobSortColumn -> JobSortColumn -> [JobSortColumn])
-> Enum JobSortColumn
forall a.
(a -> a)
-> (a -> a)
-> (Int -> a)
-> (a -> Int)
-> (a -> [a])
-> (a -> a -> [a])
-> (a -> a -> [a])
-> (a -> a -> a -> [a])
-> Enum a
$csucc :: JobSortColumn -> JobSortColumn
succ :: JobSortColumn -> JobSortColumn
$cpred :: JobSortColumn -> JobSortColumn
pred :: JobSortColumn -> JobSortColumn
$ctoEnum :: Int -> JobSortColumn
toEnum :: Int -> JobSortColumn
$cfromEnum :: JobSortColumn -> Int
fromEnum :: JobSortColumn -> Int
$cenumFrom :: JobSortColumn -> [JobSortColumn]
enumFrom :: JobSortColumn -> [JobSortColumn]
$cenumFromThen :: JobSortColumn -> JobSortColumn -> [JobSortColumn]
enumFromThen :: JobSortColumn -> JobSortColumn -> [JobSortColumn]
$cenumFromTo :: JobSortColumn -> JobSortColumn -> [JobSortColumn]
enumFromTo :: JobSortColumn -> JobSortColumn -> [JobSortColumn]
$cenumFromThenTo :: JobSortColumn -> JobSortColumn -> JobSortColumn -> [JobSortColumn]
enumFromThenTo :: JobSortColumn -> JobSortColumn -> JobSortColumn -> [JobSortColumn]
Enum, JobSortColumn -> JobSortColumn -> Bool
(JobSortColumn -> JobSortColumn -> Bool)
-> (JobSortColumn -> JobSortColumn -> Bool) -> Eq JobSortColumn
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: JobSortColumn -> JobSortColumn -> Bool
== :: JobSortColumn -> JobSortColumn -> Bool
$c/= :: JobSortColumn -> JobSortColumn -> Bool
/= :: JobSortColumn -> JobSortColumn -> Bool
Eq, Int -> JobSortColumn -> ShowS
[JobSortColumn] -> ShowS
JobSortColumn -> String
(Int -> JobSortColumn -> ShowS)
-> (JobSortColumn -> String)
-> ([JobSortColumn] -> ShowS)
-> Show JobSortColumn
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> JobSortColumn -> ShowS
showsPrec :: Int -> JobSortColumn -> ShowS
$cshow :: JobSortColumn -> String
show :: JobSortColumn -> String
$cshowList :: [JobSortColumn] -> ShowS
showList :: [JobSortColumn] -> ShowS
Show)

-- | Underlying SQL column name for a 'JobSortColumn'.
jobSortColumnName :: JobSortColumn -> Text
jobSortColumnName :: JobSortColumn -> Text
jobSortColumnName = \case
  JobSortColumn
JsId -> Text
"id"
  JobSortColumn
JsPriority -> Text
"priority"
  JobSortColumn
JsAttempts -> Text
"attempts"
  JobSortColumn
JsInsertedAt -> Text
"inserted_at"
  JobSortColumn
JsNotVisibleUntil -> Text
"not_visible_until"
  JobSortColumn
JsGroupKey -> Text
"group_key"
  JobSortColumn
JsParentId -> Text
"parent_id"
  JobSortColumn
JsLastAttemptedAt -> Text
"last_attempted_at"

-- | Sortable columns on the DLQ table. 'DlqId' is the DLQ primary key and
-- 'DlqJobId' the original job id.
data DLQSortColumn
  = -- | By @id@, the DLQ primary key.
    DlqId
  | -- | By @failed_at@.
    DlqFailedAt
  | -- | By @job_id@, the original job id.
    DlqJobId
  | -- | By @priority@.
    DlqPriority
  | -- | By @attempts@.
    DlqAttempts
  | -- | By @inserted_at@.
    DlqInsertedAt
  | -- | By @group_key@.
    DlqGroupKey
  | -- | By @parent_id@.
    DlqParentId
  | -- | By @last_attempted_at@.
    DlqLastAttemptedAt
  deriving stock (DLQSortColumn
DLQSortColumn -> DLQSortColumn -> Bounded DLQSortColumn
forall a. a -> a -> Bounded a
$cminBound :: DLQSortColumn
minBound :: DLQSortColumn
$cmaxBound :: DLQSortColumn
maxBound :: DLQSortColumn
Bounded, Int -> DLQSortColumn
DLQSortColumn -> Int
DLQSortColumn -> [DLQSortColumn]
DLQSortColumn -> DLQSortColumn
DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
DLQSortColumn -> DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
(DLQSortColumn -> DLQSortColumn)
-> (DLQSortColumn -> DLQSortColumn)
-> (Int -> DLQSortColumn)
-> (DLQSortColumn -> Int)
-> (DLQSortColumn -> [DLQSortColumn])
-> (DLQSortColumn -> DLQSortColumn -> [DLQSortColumn])
-> (DLQSortColumn -> DLQSortColumn -> [DLQSortColumn])
-> (DLQSortColumn
    -> DLQSortColumn -> DLQSortColumn -> [DLQSortColumn])
-> Enum DLQSortColumn
forall a.
(a -> a)
-> (a -> a)
-> (Int -> a)
-> (a -> Int)
-> (a -> [a])
-> (a -> a -> [a])
-> (a -> a -> [a])
-> (a -> a -> a -> [a])
-> Enum a
$csucc :: DLQSortColumn -> DLQSortColumn
succ :: DLQSortColumn -> DLQSortColumn
$cpred :: DLQSortColumn -> DLQSortColumn
pred :: DLQSortColumn -> DLQSortColumn
$ctoEnum :: Int -> DLQSortColumn
toEnum :: Int -> DLQSortColumn
$cfromEnum :: DLQSortColumn -> Int
fromEnum :: DLQSortColumn -> Int
$cenumFrom :: DLQSortColumn -> [DLQSortColumn]
enumFrom :: DLQSortColumn -> [DLQSortColumn]
$cenumFromThen :: DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
enumFromThen :: DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
$cenumFromTo :: DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
enumFromTo :: DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
$cenumFromThenTo :: DLQSortColumn -> DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
enumFromThenTo :: DLQSortColumn -> DLQSortColumn -> DLQSortColumn -> [DLQSortColumn]
Enum, DLQSortColumn -> DLQSortColumn -> Bool
(DLQSortColumn -> DLQSortColumn -> Bool)
-> (DLQSortColumn -> DLQSortColumn -> Bool) -> Eq DLQSortColumn
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: DLQSortColumn -> DLQSortColumn -> Bool
== :: DLQSortColumn -> DLQSortColumn -> Bool
$c/= :: DLQSortColumn -> DLQSortColumn -> Bool
/= :: DLQSortColumn -> DLQSortColumn -> Bool
Eq, Int -> DLQSortColumn -> ShowS
[DLQSortColumn] -> ShowS
DLQSortColumn -> String
(Int -> DLQSortColumn -> ShowS)
-> (DLQSortColumn -> String)
-> ([DLQSortColumn] -> ShowS)
-> Show DLQSortColumn
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> DLQSortColumn -> ShowS
showsPrec :: Int -> DLQSortColumn -> ShowS
$cshow :: DLQSortColumn -> String
show :: DLQSortColumn -> String
$cshowList :: [DLQSortColumn] -> ShowS
showList :: [DLQSortColumn] -> ShowS
Show)

-- | The DLQ column a sort key names.
dlqSortColumnName :: DLQSortColumn -> Text
dlqSortColumnName :: DLQSortColumn -> Text
dlqSortColumnName = \case
  DLQSortColumn
DlqId -> Text
"id"
  DLQSortColumn
DlqFailedAt -> Text
"failed_at"
  DLQSortColumn
DlqJobId -> Text
"job_id"
  DLQSortColumn
DlqPriority -> Text
"priority"
  DLQSortColumn
DlqAttempts -> Text
"attempts"
  DLQSortColumn
DlqInsertedAt -> Text
"inserted_at"
  DLQSortColumn
DlqGroupKey -> Text
"group_key"
  DLQSortColumn
DlqParentId -> Text
"parent_id"
  DLQSortColumn
DlqLastAttemptedAt -> Text
"last_attempted_at"

-- | Sortable columns on the archive table. 'ArchiveId' is the archive primary
-- key and 'ArchiveJobId' the original job id.
data ArchiveSortColumn
  = -- | By @id@, the archive primary key.
    ArchiveId
  | -- | By @completed_at@.
    ArchiveCompletedAt
  | -- | By @inserted_at@.
    ArchiveInsertedAt
  | -- | By @job_id@, the original job id.
    ArchiveJobId
  | -- | By @attempts@.
    ArchiveAttempts
  | -- | By @group_key@.
    ArchiveGroupKey
  | -- | By @parent_id@.
    ArchiveParentId
  deriving stock (ArchiveSortColumn
ArchiveSortColumn -> ArchiveSortColumn -> Bounded ArchiveSortColumn
forall a. a -> a -> Bounded a
$cminBound :: ArchiveSortColumn
minBound :: ArchiveSortColumn
$cmaxBound :: ArchiveSortColumn
maxBound :: ArchiveSortColumn
Bounded, Int -> ArchiveSortColumn
ArchiveSortColumn -> Int
ArchiveSortColumn -> [ArchiveSortColumn]
ArchiveSortColumn -> ArchiveSortColumn
ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
ArchiveSortColumn
-> ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
(ArchiveSortColumn -> ArchiveSortColumn)
-> (ArchiveSortColumn -> ArchiveSortColumn)
-> (Int -> ArchiveSortColumn)
-> (ArchiveSortColumn -> Int)
-> (ArchiveSortColumn -> [ArchiveSortColumn])
-> (ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn])
-> (ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn])
-> (ArchiveSortColumn
    -> ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn])
-> Enum ArchiveSortColumn
forall a.
(a -> a)
-> (a -> a)
-> (Int -> a)
-> (a -> Int)
-> (a -> [a])
-> (a -> a -> [a])
-> (a -> a -> [a])
-> (a -> a -> a -> [a])
-> Enum a
$csucc :: ArchiveSortColumn -> ArchiveSortColumn
succ :: ArchiveSortColumn -> ArchiveSortColumn
$cpred :: ArchiveSortColumn -> ArchiveSortColumn
pred :: ArchiveSortColumn -> ArchiveSortColumn
$ctoEnum :: Int -> ArchiveSortColumn
toEnum :: Int -> ArchiveSortColumn
$cfromEnum :: ArchiveSortColumn -> Int
fromEnum :: ArchiveSortColumn -> Int
$cenumFrom :: ArchiveSortColumn -> [ArchiveSortColumn]
enumFrom :: ArchiveSortColumn -> [ArchiveSortColumn]
$cenumFromThen :: ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
enumFromThen :: ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
$cenumFromTo :: ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
enumFromTo :: ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
$cenumFromThenTo :: ArchiveSortColumn
-> ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
enumFromThenTo :: ArchiveSortColumn
-> ArchiveSortColumn -> ArchiveSortColumn -> [ArchiveSortColumn]
Enum, ArchiveSortColumn -> ArchiveSortColumn -> Bool
(ArchiveSortColumn -> ArchiveSortColumn -> Bool)
-> (ArchiveSortColumn -> ArchiveSortColumn -> Bool)
-> Eq ArchiveSortColumn
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ArchiveSortColumn -> ArchiveSortColumn -> Bool
== :: ArchiveSortColumn -> ArchiveSortColumn -> Bool
$c/= :: ArchiveSortColumn -> ArchiveSortColumn -> Bool
/= :: ArchiveSortColumn -> ArchiveSortColumn -> Bool
Eq, Int -> ArchiveSortColumn -> ShowS
[ArchiveSortColumn] -> ShowS
ArchiveSortColumn -> String
(Int -> ArchiveSortColumn -> ShowS)
-> (ArchiveSortColumn -> String)
-> ([ArchiveSortColumn] -> ShowS)
-> Show ArchiveSortColumn
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ArchiveSortColumn -> ShowS
showsPrec :: Int -> ArchiveSortColumn -> ShowS
$cshow :: ArchiveSortColumn -> String
show :: ArchiveSortColumn -> String
$cshowList :: [ArchiveSortColumn] -> ShowS
showList :: [ArchiveSortColumn] -> ShowS
Show)

-- | The archive column a sort key names.
archiveSortColumnName :: ArchiveSortColumn -> Text
archiveSortColumnName :: ArchiveSortColumn -> Text
archiveSortColumnName = \case
  ArchiveSortColumn
ArchiveId -> Text
"id"
  ArchiveSortColumn
ArchiveCompletedAt -> Text
"completed_at"
  ArchiveSortColumn
ArchiveInsertedAt -> Text
"inserted_at"
  ArchiveSortColumn
ArchiveJobId -> Text
"job_id"
  ArchiveSortColumn
ArchiveAttempts -> Text
"attempts"
  ArchiveSortColumn
ArchiveGroupKey -> Text
"group_key"
  ArchiveSortColumn
ArchiveParentId -> Text
"parent_id"

-- | Sort direction.
data SortDir
  = -- | Ascending.
    SortAsc
  | -- | Descending.
    SortDesc
  deriving stock (SortDir
SortDir -> SortDir -> Bounded SortDir
forall a. a -> a -> Bounded a
$cminBound :: SortDir
minBound :: SortDir
$cmaxBound :: SortDir
maxBound :: SortDir
Bounded, Int -> SortDir
SortDir -> Int
SortDir -> [SortDir]
SortDir -> SortDir
SortDir -> SortDir -> [SortDir]
SortDir -> SortDir -> SortDir -> [SortDir]
(SortDir -> SortDir)
-> (SortDir -> SortDir)
-> (Int -> SortDir)
-> (SortDir -> Int)
-> (SortDir -> [SortDir])
-> (SortDir -> SortDir -> [SortDir])
-> (SortDir -> SortDir -> [SortDir])
-> (SortDir -> SortDir -> SortDir -> [SortDir])
-> Enum SortDir
forall a.
(a -> a)
-> (a -> a)
-> (Int -> a)
-> (a -> Int)
-> (a -> [a])
-> (a -> a -> [a])
-> (a -> a -> [a])
-> (a -> a -> a -> [a])
-> Enum a
$csucc :: SortDir -> SortDir
succ :: SortDir -> SortDir
$cpred :: SortDir -> SortDir
pred :: SortDir -> SortDir
$ctoEnum :: Int -> SortDir
toEnum :: Int -> SortDir
$cfromEnum :: SortDir -> Int
fromEnum :: SortDir -> Int
$cenumFrom :: SortDir -> [SortDir]
enumFrom :: SortDir -> [SortDir]
$cenumFromThen :: SortDir -> SortDir -> [SortDir]
enumFromThen :: SortDir -> SortDir -> [SortDir]
$cenumFromTo :: SortDir -> SortDir -> [SortDir]
enumFromTo :: SortDir -> SortDir -> [SortDir]
$cenumFromThenTo :: SortDir -> SortDir -> SortDir -> [SortDir]
enumFromThenTo :: SortDir -> SortDir -> SortDir -> [SortDir]
Enum, SortDir -> SortDir -> Bool
(SortDir -> SortDir -> Bool)
-> (SortDir -> SortDir -> Bool) -> Eq SortDir
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: SortDir -> SortDir -> Bool
== :: SortDir -> SortDir -> Bool
$c/= :: SortDir -> SortDir -> Bool
/= :: SortDir -> SortDir -> Bool
Eq, Int -> SortDir -> ShowS
[SortDir] -> ShowS
SortDir -> String
(Int -> SortDir -> ShowS)
-> (SortDir -> String) -> ([SortDir] -> ShowS) -> Show SortDir
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SortDir -> ShowS
showsPrec :: Int -> SortDir -> ShowS
$cshow :: SortDir -> String
show :: SortDir -> String
$cshowList :: [SortDir] -> ShowS
showList :: [SortDir] -> ShowS
Show)

-- | A sort direction as SQL.
sortDirSql :: SortDir -> Text
sortDirSql :: SortDir -> Text
sortDirSql SortDir
SortAsc = Text
"ASC"
sortDirSql SortDir
SortDesc = Text
"DESC"

-- | A sort direction as a query parameter.
sortDirName :: SortDir -> Text
sortDirName :: SortDir -> Text
sortDirName = Text -> Text
T.toLower (Text -> Text) -> (SortDir -> Text) -> SortDir -> Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SortDir -> Text
sortDirSql

-- | A throttle-deferred job with a live marker, still parked. Claiming clears
-- the marker.
throttledPredicateSQL :: Text
throttledPredicateSQL :: Text
throttledPredicateSQL =
  Text
"throttled_until > NOW() AND not_visible_until > NOW()"

-- | The derived job status. Its string values match 'Arbiter.Core.Job.Status.jobStatusToText'.
-- An exhausted job is visible but out of attempts, awaiting the reaper's DLQ sweep.
jobStatusCaseSQL :: Text
jobStatusCaseSQL :: Text
jobStatusCaseSQL =
  [text|
    CASE
      WHEN cancel_requested_at IS NOT NULL THEN 'cancelled'
      WHEN suspended THEN 'suspended'
      WHEN ${throttledPredicateSQL} THEN 'throttled'
      WHEN claimed_by IS NULL AND attempts > 0
           AND not_visible_until IS NOT NULL AND not_visible_until > NOW() THEN 'backoff'
      WHEN attempts > 0 AND not_visible_until IS NOT NULL AND not_visible_until > NOW() THEN 'in_flight'
      WHEN not_visible_until IS NOT NULL AND not_visible_until > NOW() THEN 'scheduled'
      WHEN ${spent} THEN 'exhausted'
      ELSE 'ready'
    END
  |]
  where
    spent :: Text
spent = Text -> Text
attemptsSpentSQL Text
""

-- | The rows a claim considers, over a row alias: unsuspended, uncancelled, visible,
-- and within the attempt budget.
claimablePred :: Text -> Text
claimablePred :: Text -> Text
claimablePred Text
alias =
  let attemptsLeft :: Text
attemptsLeft = Text -> Text
attemptsLeftSQL (Text
alias Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
".")
   in [text|
        NOT ${alias}.suspended
        AND ${alias}.cancel_requested_at IS NULL
        AND (${alias}.not_visible_until IS NULL OR ${alias}.not_visible_until <= NOW())
        AND ${attemptsLeft}
      |]

-- | The job read columns plus the derived @status@ column, aliased @job@, for filtering.
jobsWithStatusSubquery :: SchemaName -> TableName -> Text
jobsWithStatusSubquery :: Text -> Text -> Text
jobsWithStatusSubquery Text
schemaName Text
tableName =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
   in [text|(SELECT ${jobReadColumns}, ${jobStatusCaseSQL} AS status FROM ${tbl}) job|]

-- | List filtered jobs without the derived status.
listJobsFilteredSQL :: SchemaName -> TableName -> Query () -> Text -> Int64 -> Int64 -> Query (JobRead (Stored payload))
listJobsFilteredSQL :: forall payload.
Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (JobRead (Stored payload))
listJobsFilteredSQL Text
schemaName Text
tableName Query ()
whereFrag Text
orderBy Int64
limit Int64
offset =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
   in RowCodec (JobRead (Stored payload))
-> Query () -> Query (JobRead (Stored payload))
forall a. RowCodec a -> Query () -> Query a
rows
        (Text -> RowCodec (JobRead (Stored payload))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName)
        [sql|
          SELECT ${jobReadColumns}
          FROM ${tbl}
          ${whereFrag}
          ORDER BY ${orderBy} LIMIT #{limit :: CInt8} OFFSET #{offset :: CInt8}
        |]

-- | List filtered jobs with @status@ as a trailing column. The caller attaches the row decoder.
listJobsWithStatusSQL :: SchemaName -> TableName -> Query () -> Text -> Int64 -> Int64 -> Query ()
listJobsWithStatusSQL :: Text -> Text -> Query () -> Text -> Int64 -> Int64 -> Query ()
listJobsWithStatusSQL Text
schemaName Text
tableName Query ()
whereFrag Text
orderBy Int64
limit Int64
offset =
  let sub :: Text
sub = Text -> Text -> Text
jobsWithStatusSubquery Text
schemaName Text
tableName
   in [sql|
        SELECT * FROM ${sub}
        ${whereFrag}
        ORDER BY ${orderBy} LIMIT #{limit :: CInt8} OFFSET #{offset :: CInt8}
      |]

-- | Count filtered jobs through the status subquery.
countJobsFilteredSQL :: SchemaName -> TableName -> Query () -> Query Int64
countJobsFilteredSQL :: Text -> Text -> Query () -> Query Int64
countJobsFilteredSQL Text
schemaName Text
tableName Query ()
whereFrag =
  let sub :: Text
sub = Text -> Text -> Text
jobsWithStatusSubquery Text
schemaName Text
tableName
   in Text -> Query () -> Query Int64
countWhereSQL Text
sub Query ()
whereFrag

-- | Fetch a single job by id with its derived @status@ trailing column. The caller attaches the row decoder.
getJobByIdWithStatusSQL :: SchemaName -> TableName -> Int64 -> Query ()
getJobByIdWithStatusSQL :: Text -> Text -> Int64 -> Query ()
getJobByIdWithStatusSQL Text
schemaName Text
tableName Int64
jobId =
  let sub :: Text
sub = Text -> Text -> Text
jobsWithStatusSubquery Text
schemaName Text
tableName
   in [sql|SELECT * FROM ${sub} WHERE id = #{jobId :: CInt8}|]

-- | How NULLs in a sort column should order relative to non-NULL values.
data NullsBehavior
  = -- | Column is NOT NULL. Emit no NULLS clause.
    NullsNotApplicable
  | -- | NULL means absent. Always sorts last.
    NullsAsAbsent
  | -- | NULL is the column's minimum (e.g. @not_visible_until IS NULL@ means visible now).
    -- NULLS FIRST when ASC, NULLS LAST when DESC.
    NullsAsMinimum

-- | The @NULLS@ placement for a column's null semantics and direction.
nullsClause :: NullsBehavior -> SortDir -> Text
nullsClause :: NullsBehavior -> SortDir -> Text
nullsClause NullsBehavior
NullsNotApplicable SortDir
_ = Text
""
nullsClause NullsBehavior
NullsAsAbsent SortDir
_ = Text
" NULLS LAST"
nullsClause NullsBehavior
NullsAsMinimum SortDir
SortAsc = Text
" NULLS FIRST"
nullsClause NullsBehavior
NullsAsMinimum SortDir
SortDesc = Text
" NULLS LAST"

-- | How nulls sort for a job column.
jobColumnNulls :: JobSortColumn -> NullsBehavior
jobColumnNulls :: JobSortColumn -> NullsBehavior
jobColumnNulls = \case
  JobSortColumn
JsId -> NullsBehavior
NullsNotApplicable
  JobSortColumn
JsPriority -> NullsBehavior
NullsNotApplicable
  JobSortColumn
JsAttempts -> NullsBehavior
NullsNotApplicable
  JobSortColumn
JsInsertedAt -> NullsBehavior
NullsNotApplicable
  JobSortColumn
JsNotVisibleUntil -> NullsBehavior
NullsAsMinimum
  JobSortColumn
JsGroupKey -> NullsBehavior
NullsAsAbsent
  JobSortColumn
JsParentId -> NullsBehavior
NullsAsAbsent
  JobSortColumn
JsLastAttemptedAt -> NullsBehavior
NullsAsMinimum

-- | How nulls sort for a DLQ column.
dlqColumnNulls :: DLQSortColumn -> NullsBehavior
dlqColumnNulls :: DLQSortColumn -> NullsBehavior
dlqColumnNulls = \case
  DLQSortColumn
DlqId -> NullsBehavior
NullsNotApplicable
  DLQSortColumn
DlqFailedAt -> NullsBehavior
NullsNotApplicable
  DLQSortColumn
DlqJobId -> NullsBehavior
NullsNotApplicable
  DLQSortColumn
DlqInsertedAt -> NullsBehavior
NullsNotApplicable
  DLQSortColumn
DlqPriority -> NullsBehavior
NullsNotApplicable
  DLQSortColumn
DlqAttempts -> NullsBehavior
NullsNotApplicable
  DLQSortColumn
DlqGroupKey -> NullsBehavior
NullsAsAbsent
  DLQSortColumn
DlqParentId -> NullsBehavior
NullsAsAbsent
  DLQSortColumn
DlqLastAttemptedAt -> NullsBehavior
NullsAsMinimum

archiveColumnNulls :: ArchiveSortColumn -> NullsBehavior
archiveColumnNulls :: ArchiveSortColumn -> NullsBehavior
archiveColumnNulls = \case
  ArchiveSortColumn
ArchiveId -> NullsBehavior
NullsNotApplicable
  ArchiveSortColumn
ArchiveCompletedAt -> NullsBehavior
NullsNotApplicable
  ArchiveSortColumn
ArchiveInsertedAt -> NullsBehavior
NullsNotApplicable
  ArchiveSortColumn
ArchiveJobId -> NullsBehavior
NullsNotApplicable
  ArchiveSortColumn
ArchiveAttempts -> NullsBehavior
NullsNotApplicable
  ArchiveSortColumn
ArchiveGroupKey -> NullsBehavior
NullsAsAbsent
  ArchiveSortColumn
ArchiveParentId -> NullsBehavior
NullsAsAbsent

-- | Build an ORDER BY clause from a typed sort spec. @nullsFn@ places NULLs per
-- column. A stable @id@ tie-breaker in the primary sort's direction follows any
-- sort column other than @idCol@.
buildOrderBy :: (Eq a) => (a -> Text) -> (a -> NullsBehavior) -> a -> a -> Maybe a -> Maybe SortDir -> Text
buildOrderBy :: forall a.
Eq a =>
(a -> Text)
-> (a -> NullsBehavior)
-> a
-> a
-> Maybe a
-> Maybe SortDir
-> Text
buildOrderBy a -> Text
nameFn a -> NullsBehavior
nullsFn a
defCol a
idCol Maybe a
mCol Maybe SortDir
mDir =
  let sortCol :: a
sortCol = a -> Maybe a -> a
forall a. a -> Maybe a -> a
fromMaybe a
defCol Maybe a
mCol
      dir :: SortDir
dir = SortDir -> Maybe SortDir -> SortDir
forall a. a -> Maybe a -> a
fromMaybe SortDir
SortDesc Maybe SortDir
mDir
      columnName :: Text
columnName = a -> Text
nameFn a
sortCol
      dirText :: Text
dirText = SortDir -> Text
sortDirSql SortDir
dir
      nulls :: Text
nulls = NullsBehavior -> SortDir -> Text
nullsClause (a -> NullsBehavior
nullsFn a
sortCol) SortDir
dir
      tieBreaker :: Text
tieBreaker = Bool -> Text -> Text
forall m. Monoid m => Bool -> m -> m
mwhen (a
sortCol a -> a -> Bool
forall a. Eq a => a -> a -> Bool
/= a
idCol) [text|, id ${dirText}|]
   in [text|${columnName} ${dirText}${nulls}${tieBreaker}|]

-- | ORDER BY for the jobs table. Defaults to @id DESC@.
buildJobsOrderBy :: Maybe JobSortColumn -> Maybe SortDir -> Text
buildJobsOrderBy :: Maybe JobSortColumn -> Maybe SortDir -> Text
buildJobsOrderBy = (JobSortColumn -> Text)
-> (JobSortColumn -> NullsBehavior)
-> JobSortColumn
-> JobSortColumn
-> Maybe JobSortColumn
-> Maybe SortDir
-> Text
forall a.
Eq a =>
(a -> Text)
-> (a -> NullsBehavior)
-> a
-> a
-> Maybe a
-> Maybe SortDir
-> Text
buildOrderBy JobSortColumn -> Text
jobSortColumnName JobSortColumn -> NullsBehavior
jobColumnNulls JobSortColumn
JsId JobSortColumn
JsId

-- | ORDER BY for the DLQ table. Defaults to @failed_at DESC@.
buildDLQOrderBy :: Maybe DLQSortColumn -> Maybe SortDir -> Text
buildDLQOrderBy :: Maybe DLQSortColumn -> Maybe SortDir -> Text
buildDLQOrderBy = (DLQSortColumn -> Text)
-> (DLQSortColumn -> NullsBehavior)
-> DLQSortColumn
-> DLQSortColumn
-> Maybe DLQSortColumn
-> Maybe SortDir
-> Text
forall a.
Eq a =>
(a -> Text)
-> (a -> NullsBehavior)
-> a
-> a
-> Maybe a
-> Maybe SortDir
-> Text
buildOrderBy DLQSortColumn -> Text
dlqSortColumnName DLQSortColumn -> NullsBehavior
dlqColumnNulls DLQSortColumn
DlqFailedAt DLQSortColumn
DlqId

-- | ORDER BY for the archive table. Defaults to @completed_at DESC@.
buildArchiveOrderBy :: Maybe ArchiveSortColumn -> Maybe SortDir -> Text
buildArchiveOrderBy :: Maybe ArchiveSortColumn -> Maybe SortDir -> Text
buildArchiveOrderBy = (ArchiveSortColumn -> Text)
-> (ArchiveSortColumn -> NullsBehavior)
-> ArchiveSortColumn
-> ArchiveSortColumn
-> Maybe ArchiveSortColumn
-> Maybe SortDir
-> Text
forall a.
Eq a =>
(a -> Text)
-> (a -> NullsBehavior)
-> a
-> a
-> Maybe a
-> Maybe SortDir
-> Text
buildOrderBy ArchiveSortColumn -> Text
archiveSortColumnName ArchiveSortColumn -> NullsBehavior
archiveColumnNulls ArchiveSortColumn
ArchiveCompletedAt ArchiveSortColumn
ArchiveId

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

filterToClause :: FilterTable -> JobFilter -> Query ()
filterToClause :: FilterTable -> JobFilter -> Query ()
filterToClause FilterTable
_ (FilterGroupKey Text
key) = [sql|group_key = #{key :: CText}|]
filterToClause FilterTable
_ (FilterParentId Int64
pid) = [sql|parent_id = #{pid :: CInt8}|]
filterToClause FilterTable
_ JobFilter
FilterRootsOnly = Text -> Query ()
raw Text
"parent_id IS NULL"
filterToClause FilterTable
MainTable (FilterStatus JobStatus
status) = [sql|status = #{statusText :: CText}|]
  where
    statusText :: Text
statusText = JobStatus -> Text
jobStatusToText JobStatus
status
filterToClause FilterTable
_ (FilterStatus JobStatus
_) = Query ()
matchNothing
filterToClause FilterTable
_ (FilterId Int64
jobId) = [sql|id = #{jobId :: CInt8}|]
filterToClause FilterTable
MainTable (FilterJobId Int64
jobId) = [sql|id = #{jobId :: CInt8}|]
filterToClause FilterTable
_ (FilterJobId Int64
jobId) = [sql|job_id = #{jobId :: CInt8}|]
filterToClause FilterTable
_ (FilterClaimedBy UUID
workerId) = [sql|claimed_by = #{workerId :: CUuid}|]
filterToClause FilterTable
_ (FilterKind Text
kind) = [sql|kind = #{kind :: CText}|]
filterToClause FilterTable
_ (FilterRateLimitPrefix Text
prefix) = [sql|rate_limit_prefix = #{prefix :: CText}|]
filterToClause FilterTable
_ (FilterConcurrencyPrefix Text
prefix) = [sql|concurrency_prefix = #{prefix :: CText}|]
filterToClause FilterTable
_ (FilterInsertedAfter UTCTime
time) = [sql|inserted_at >= #{time :: CTimestamptz}|]
filterToClause FilterTable
_ (FilterInsertedBefore UTCTime
time) = [sql|inserted_at < #{time :: CTimestamptz}|]
filterToClause FilterTable
ArchiveTable (FilterCompletedAfter UTCTime
time) = [sql|completed_at >= #{time :: CTimestamptz}|]
filterToClause FilterTable
ArchiveTable (FilterCompletedBefore UTCTime
time) = [sql|completed_at < #{time :: CTimestamptz}|]
filterToClause FilterTable
_ (FilterCompletedAfter UTCTime
_) = Query ()
matchNothing
filterToClause FilterTable
_ (FilterCompletedBefore UTCTime
_) = Query ()
matchNothing
filterToClause FilterTable
_ (FilterPayloadText Text
needle) = Text -> Text -> Query ()
containsClause Text
"payload::text" Text
needle
filterToClause FilterTable
_ (FilterErrorText Text
needle) = Text -> Text -> Query ()
containsClause Text
"last_error" Text
needle

-- | A predicate on a column the table does not have.
matchNothing :: Query ()
matchNothing :: Query ()
matchNothing = Text -> Query ()
raw Text
"FALSE"

-- | A case-insensitive match of any @column@ text that contains @needle@ literally.
containsClause :: Text -> Text -> Query ()
containsClause :: Text -> Text -> Query ()
containsClause Text
column Text
needle = [sql|${column} ILIKE #{pat :: CText} ESCAPE '\' |]
  where
    pat :: Text
pat = Text
"%" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
likeEscape Text
needle Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"%"

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

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

-- | Count the rows of @src@ under a dynamic WHERE.
countWhereSQL :: Text -> Query () -> Query Int64
countWhereSQL :: Text -> Query () -> Query Int64
countWhereSQL Text
src Query ()
whereFrag = [sql|SELECT COUNT(*) AS @{count :: CInt8} FROM ${src} ${whereFrag}|]

-- | The job read columns, in codec order, for SELECT and RETURNING.
jobReadColumns :: Text
jobReadColumns :: Text
jobReadColumns = [Text] -> Text
joinColumns [Text]
jobReadColumnNames

-- | The job read column names, in codec order.
jobReadColumnNames :: [Text]
jobReadColumnNames :: [Text]
jobReadColumnNames = RowCodec (JobRead (Stored (ZonkAny 0))) -> [Text]
forall a. RowCodec a -> [Text]
codecColumns (Text -> RowCodec (JobRead (Stored (ZonkAny 0)))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
"")

-- | The job read columns except @id@. The archive INSERT copies them with the main
-- table's @id@ as @job_id@.
jobColsExceptId :: Text
jobColsExceptId :: Text
jobColsExceptId = [Text] -> Text
joinColumns (Int -> [Text] -> [Text]
forall a. Int -> [a] -> [a]
drop Int
1 [Text]
jobReadColumnNames)

-- | Job columns carried through a DLQ round-trip. The read columns except @id@ and
-- @last_error@, plus write-only @rate_limit_cost@.
dlqCarriedCols :: Text
dlqCarriedCols :: Text
dlqCarriedCols = [Text] -> Text
joinColumns ((Text -> Bool) -> [Text] -> [Text]
forall a. (a -> Bool) -> [a] -> [a]
filter (Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
/= Text
"last_error") (Int -> [Text] -> [Text]
forall a. Int -> [a] -> [a]
drop Int
1 [Text]
jobReadColumnNames) [Text] -> [Text] -> [Text]
forall a. Semigroup a => a -> a -> a
<> [Text
"rate_limit_cost"])

-- | Job columns a DLQ retry carries back to the main table. The retry re-arms the rest.
requeuedColumnNames :: [Text]
requeuedColumnNames :: [Text]
requeuedColumnNames =
  [ Text
"payload"
  , Text
"group_key"
  , Text
"priority"
  , Text
"max_attempts"
  , Text
"parent_id"
  , Text
"parent_state"
  , Text
"traceparent"
  , Text
"tracestate"
  , Text
"archive_for"
  , Text
"kind"
  , Text
"rate_limit_key"
  , Text
"rate_limit_prefix"
  , Text
"concurrency_key"
  , Text
"concurrency_prefix"
  , Text
"rate_limit_cost"
  ]

-- | The job columns a DLQ retry carries back to the main table, as a column list.
requeuedCols :: Text
requeuedCols :: Text
requeuedCols = [Text] -> Text
joinColumns [Text]
requeuedColumnNames

-- | 'requeuedCols' as select expressions, with the attempt limit clamped. Each
-- column in @edited@ is read through @reader@.
requeuedValsEditing :: [Text] -> (Text -> Text) -> Text
requeuedValsEditing :: [Text] -> (Text -> Text) -> Text
requeuedValsEditing [Text]
edited Text -> Text
reader = [Text] -> (Text -> Text) -> [Text] -> Text
valsEditing [Text]
edited Text -> Text
reader [Text]
requeuedColumnNames

-- | 'requeuedColumnNames' for an archive re-enqueue, without the parent link.
enqueuedAgainColumnNames :: [Text]
enqueuedAgainColumnNames :: [Text]
enqueuedAgainColumnNames = (Text -> Bool) -> [Text] -> [Text]
forall a. (a -> Bool) -> [a] -> [a]
filter (Text -> [Text] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`notElem` [Text
"parent_id", Text
"parent_state"]) [Text]
requeuedColumnNames

-- | The job columns an archive re-enqueue carries, as a column list. No parent link.
enqueuedAgainCols :: Text
enqueuedAgainCols :: Text
enqueuedAgainCols = [Text] -> Text
joinColumns [Text]
enqueuedAgainColumnNames

-- | 'enqueuedAgainCols' as select expressions, with the attempt limit clamped. Each
-- column in @edited@ is read through @reader@.
enqueuedAgainValsEditing :: [Text] -> (Text -> Text) -> Text
enqueuedAgainValsEditing :: [Text] -> (Text -> Text) -> Text
enqueuedAgainValsEditing [Text]
edited Text -> Text
reader = [Text] -> (Text -> Text) -> [Text] -> Text
valsEditing [Text]
edited Text -> Text
reader [Text]
enqueuedAgainColumnNames

valsEditing :: [Text] -> (Text -> Text) -> [Text] -> Text
valsEditing :: [Text] -> (Text -> Text) -> [Text] -> Text
valsEditing [Text]
edited Text -> Text
reader = [Text] -> Text
joinColumns ([Text] -> Text) -> ([Text] -> [Text]) -> [Text] -> Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map Text -> Text
pick
  where
    pick :: Text -> Text
pick Text
column
      | Text
column Text -> [Text] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` [Text]
edited = Text -> Text
reader Text
column
      | Text
column Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
== Text
"max_attempts" = Text
clampedMaxAttempts
      | Bool
otherwise = Text
column
    clampedMaxAttempts :: Text
clampedMaxAttempts = [text|GREATEST(COALESCE(max_attempts, ${defaultMaxAttemptsSQL}), ${minMaxAttemptsSQL})|]

-- | @DO UPDATE SET@ body for a replace-dedup upsert. Copies each writable column
-- from the excluded row, then re-arms the replaced job for a fresh run.
dedupUpdateSet :: Text -> Text
dedupUpdateSet :: Text -> Text
dedupUpdateSet Text
tbl =
  let copied :: Text
copied = Text -> [Text] -> Text
T.intercalate Text
", " ((Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map Text -> Text
excludedAssignment ((Text -> Bool) -> [Text] -> [Text]
forall a. (a -> Bool) -> [a] -> [a]
filter (Text -> [Text] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`notElem` [Text]
dedupResetColumns) [Text]
writeColumnNames))
   in [text|
    ${copied},
    attempts = 0,
    claim_seq = ${tbl}.claim_seq + 1,
    last_error = NULL,
    updated_at = NOW(),
    throttled_until = NULL,
    last_attempted_at = NULL,
    claimed_by = NULL
  |]

-- | Writable columns a replace re-arms.
dedupResetColumns :: [Text]
dedupResetColumns :: [Text]
dedupResetColumns = [Text
"dedup_key", Text
"attempts", Text
"last_error"]

-- | An existing row is replaceable when idle, unflagged, childless and same-parent.
replaceableGuard :: Text -> Text -> Text
replaceableGuard :: Text -> Text -> Text
replaceableGuard Text
tbl Text
dlqTbl =
  [text|
    (${tbl}.attempts = 0
      OR ${tbl}.claimed_by IS NULL
      OR ${tbl}.not_visible_until IS NULL
      OR ${tbl}.not_visible_until <= NOW())
      AND ${tbl}.cancel_requested_at IS NULL
      AND ${tbl}.parent_id IS NOT DISTINCT FROM EXCLUDED.parent_id
      AND NOT EXISTS (SELECT 1 FROM ${tbl} child WHERE child.parent_id = ${tbl}.id)
      AND NOT EXISTS (SELECT 1 FROM ${dlqTbl} dlq_child WHERE dlq_child.parent_id = ${tbl}.id)
  |]

-- | Insert a job. The write fragment carries the column list and parameters.
insertJobSQL :: SchemaName -> TableName -> Query () -> Query (JobRead (Stored payload))
insertJobSQL :: forall payload.
Text -> Text -> Query () -> Query (JobRead (Stored payload))
insertJobSQL Text
schemaName Text
tableName Query ()
valuesFrag =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
   in RowCodec (JobRead (Stored payload))
-> Query () -> Query (JobRead (Stored payload))
forall a. RowCodec a -> Query () -> Query a
rows
        (Text -> RowCodec (JobRead (Stored payload))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName)
        [sql|
          INSERT INTO ${tbl} ${valuesFrag}
          ON CONFLICT (dedup_key) WHERE dedup_key IS NOT NULL DO NOTHING
          RETURNING ${jobReadColumns}
        |]

-- | Insert under the replace dedup strategy. Replaces an existing job that is idle
-- and childless in the main queue and the DLQ. @ON CONFLICT DO UPDATE@ fires the
-- groups UPDATE trigger, which maintains a cross-group move.
insertJobReplaceSQL :: SchemaName -> TableName -> Query () -> Query (JobRead (Stored payload))
insertJobReplaceSQL :: forall payload.
Text -> Text -> Query () -> Query (JobRead (Stored payload))
insertJobReplaceSQL Text
schemaName Text
tableName Query ()
valuesFrag =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      guard :: Text
guard = Text -> Text -> Text
replaceableGuard Text
tbl Text
dlqTbl
      dedupSet :: Text
dedupSet = Text -> Text
dedupUpdateSet Text
tbl
   in RowCodec (JobRead (Stored payload))
-> Query () -> Query (JobRead (Stored payload))
forall a. RowCodec a -> Query () -> Query a
rows
        (Text -> RowCodec (JobRead (Stored payload))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName)
        [sql|
          INSERT INTO ${tbl} ${valuesFrag}
          ON CONFLICT (dedup_key) WHERE dedup_key IS NOT NULL DO UPDATE SET
            ${dedupSet}
          WHERE ${guard}
          RETURNING ${jobReadColumns}
        |]

-- | Batch insert over @unnest@ed parallel arrays. An ignore-dedup job is skipped on
-- conflict. A replace-dedup job updates an idle existing row. A row whose parent
-- is absent is skipped.
insertJobsBatchSQL :: SchemaName -> TableName -> Query () -> Query (JobRead (Stored payload))
insertJobsBatchSQL :: forall payload.
Text -> Text -> Query () -> Query (JobRead (Stored payload))
insertJobsBatchSQL Text
schemaName Text
tableName Query ()
batchSrc =
  RowCodec (JobRead (Stored payload))
-> Query () -> Query (JobRead (Stored payload))
forall a. RowCodec a -> Query () -> Query a
rows (Text -> RowCodec (JobRead (Stored payload))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName) (Text -> Text -> Query () -> Text -> Query ()
insertJobsBatchBase Text
schemaName Text
tableName Query ()
batchSrc [text|RETURNING ${jobReadColumns}|])

-- | 'insertJobsBatchSQL' with no @RETURNING@.
insertJobsBatchSQL_ :: SchemaName -> TableName -> Query () -> Query ()
insertJobsBatchSQL_ :: Text -> Text -> Query () -> Query ()
insertJobsBatchSQL_ Text
schemaName Text
tableName Query ()
batchSrc =
  Text -> Text -> Query () -> Text -> Query ()
insertJobsBatchBase Text
schemaName Text
tableName Query ()
batchSrc Text
""

-- | Batch insert with dedup and replaceable-job handling, plus a caller's @RETURNING@.
insertJobsBatchBase :: SchemaName -> TableName -> Query () -> Text -> Query ()
insertJobsBatchBase :: Text -> Text -> Query () -> Text -> Query ()
insertJobsBatchBase Text
schemaName Text
tableName Query ()
batchSrc Text
returning =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      guard :: Text
guard = Text -> Text -> Text
replaceableGuard Text
tbl Text
dlqTbl
      dedupSet :: Text
dedupSet = Text -> Text
dedupUpdateSet Text
tbl
   in [sql|
        INSERT INTO ${tbl} ${batchSrc}
        WHERE (src.parent_id IS NULL
            OR EXISTS (SELECT 1 FROM ${tbl} parent WHERE parent.id = src.parent_id))
        ON CONFLICT (dedup_key) WHERE dedup_key IS NOT NULL DO UPDATE SET
          ${dedupSet}
        WHERE EXCLUDED.dedup_strategy = 'replace'
          AND ${guard}
        ${returning}
      |]

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

-- | Fetch a job by id.
getJobByIdSQL :: SchemaName -> TableName -> Int64 -> Query (JobRead (Stored payload))
getJobByIdSQL :: forall payload.
Text -> Text -> Int64 -> Query (JobRead (Stored payload))
getJobByIdSQL Text
schemaName Text
tableName Int64
jobId =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
   in RowCodec (JobRead (Stored payload))
-> Query () -> Query (JobRead (Stored payload))
forall a. RowCodec a -> Query () -> Query a
rows
        (Text -> RowCodec (JobRead (Stored payload))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName)
        [sql|
          SELECT ${jobReadColumns}
          FROM ${tbl}
          WHERE id = #{jobId :: CInt8}
        |]

-- | Fetch a job by its dedup key. The partial unique index guarantees at most one row.
getJobByDedupKeySQL :: SchemaName -> TableName -> Text -> Query (JobRead (Stored payload))
getJobByDedupKeySQL :: forall payload.
Text -> Text -> Text -> Query (JobRead (Stored payload))
getJobByDedupKeySQL Text
schemaName Text
tableName Text
key =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
   in RowCodec (JobRead (Stored payload))
-> Query () -> Query (JobRead (Stored payload))
forall a. RowCodec a -> Query () -> Query a
rows
        (Text -> RowCodec (JobRead (Stored payload))
forall payload. Text -> RowCodec (JobRead (Stored payload))
jobRowCodec Text
tableName)
        [sql|
          SELECT ${jobReadColumns}
          FROM ${tbl}
          WHERE dedup_key = #{key :: CText}
        |]

-- | Delete a job by id. Refuses one with children, which 'Arbiter.Core.Sql.Tree.cancelJobCascadeSQL' takes.
-- A deleted child with no siblings left resumes its parent for a completion round.
cancelJobSQL :: SchemaName -> TableName -> Int64 -> Query Int64
cancelJobSQL :: Text -> Text -> Int64 -> Query Int64
cancelJobSQL Text
schemaName Text
tableName Int64
jobId =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      wakeParent :: Query ()
wakeParent = Text -> Text -> Query ()
wakeParentCte Text
tbl Text
"cancel"
      groupLocks :: Text
groupLocks = Text -> Text -> Text -> [Text] -> Text
settleGroupLocksCte Text
schemaName Text
tableName Text
"cancel" [Text
"wake_parent"]
   in [sql|
        WITH cancel AS (
          DELETE FROM ${tbl}
          WHERE id = #{jobId :: CInt8}
            AND NOT EXISTS (SELECT 1 FROM ${tbl} child WHERE child.parent_id = #{jobId :: CInt8})
          RETURNING id, parent_id, group_key
        ),
        ${wakeParent},
        ${groupLocks}
        SELECT (SELECT count(*) FROM cancel) AS @{result :: CInt8}
        FROM (SELECT count(*) FROM group_locks) held
      |]

-- | @UNION ALL@ of @body@ over each job table, passing its raw name and schema-qualified reference.
unionAllOverQueueTables :: SchemaName -> [TableName] -> (TableName -> Text -> Text) -> Text
unionAllOverQueueTables :: Text -> [Text] -> (Text -> Text -> Text) -> Text
unionAllOverQueueTables Text
schemaName [Text]
tableNames Text -> Text -> Text
body =
  Text -> [Text] -> Text
T.intercalate Text
" UNION ALL " ((Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (\Text
tableName -> Text -> Text -> Text
body Text
tableName (Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName)) [Text]
tableNames)