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

-- | Internal to the arbiter packages. Not covered by the PVP.
--
-- DLQ SQL templates.
module Arbiter.Core.Sql.DLQ
  ( DLQMove (..)
  , moveToDLQSQL
  , selectExhaustedJobsSQL
  , retryFromDLQSQL
  , dlqRetryParentSQL
  , lockDLQRetryParentSQL
  , lockDLQRetryGroupsSQL
  , dlqJobExistsSQL
  , moveToDLQBatchSQL
  , deleteDLQJobsBatchSQL
  , selectExpiredDLQSQL
  , dlqParentIdsSQL
  , cascadeChildrenToDLQSQL
  , countDLQChildrenBatchSQL
  , listDLQFilteredSQL
  , countDLQFilteredSQL
  , dlqIdsFilteredSQL
  ) where

import Data.Int (Int64)
import Data.Text (Text)
import Data.Time (UTCTime)
import NeatInterpolation (text)

import Arbiter.Core.Codec (codecColumns, dlqRowCodec, jobRowCodec, joinColumns)
import Arbiter.Core.Job.Schema (SchemaName, TableName, jobQueueDLQTable, jobQueueGroupsTable, jobQueueTable)
import Arbiter.Core.Job.Types (JobRead, Stored)
import Arbiter.Core.Sql.Insert (RowEdit (..), editJoin)
import Arbiter.Core.Sql.Jobs (countWhereSQL, dlqCarriedCols, jobReadColumns, requeuedCols, requeuedValsEditing)
import Arbiter.Core.Sql.QQ (sql)
import Arbiter.Core.Sql.Query (Query, mwhen, rows)
import Arbiter.Core.Sql.Tree (childDescendantsCte, lockedByIdsCte)
import Arbiter.Core.SqlLiterals (attemptsSpentSQL, intLiteral, lockGroupKeyOnConflictSQL)

-- | The DLQ read columns, in codec order. The DLQ uses @job_id@ for the main-table @id@.
allDLQColumns :: Text
allDLQColumns :: Text
allDLQColumns = [Text] -> Text
joinColumns (RowCodec (Int64, UTCTime, JobRead (Stored (ZonkAny 0))) -> [Text]
forall a. RowCodec a -> [Text]
codecColumns (Text -> RowCodec (Int64, UTCTime, JobRead (Stored (ZonkAny 0)))
forall payload.
Text -> RowCodec (Int64, UTCTime, JobRead (Stored payload))
dlqRowCodec Text
""))

-- | Whether a DLQ move re-checks the attempt budget it was selected on.
data DLQMove
  = -- | Move the job without a re-check.
    MoveNow
  | -- | Move the job only while it is still unsuspended, uncancelled, visible and out of attempts.
    MoveIfExhausted
  deriving stock (DLQMove -> DLQMove -> Bool
(DLQMove -> DLQMove -> Bool)
-> (DLQMove -> DLQMove -> Bool) -> Eq DLQMove
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: DLQMove -> DLQMove -> Bool
== :: DLQMove -> DLQMove -> Bool
$c/= :: DLQMove -> DLQMove -> Bool
/= :: DLQMove -> DLQMove -> Bool
Eq, Int -> DLQMove -> ShowS
[DLQMove] -> ShowS
DLQMove -> String
(Int -> DLQMove -> ShowS)
-> (DLQMove -> String) -> ([DLQMove] -> ShowS) -> Show DLQMove
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> DLQMove -> ShowS
showsPrec :: Int -> DLQMove -> ShowS
$cshow :: DLQMove -> String
show :: DLQMove -> String
$cshowList :: [DLQMove] -> ShowS
showList :: [DLQMove] -> ShowS
Show)

-- | The sweep's predicate. Unsuspended, uncancelled, visible, and out of attempts.
sweepableGuard :: Text
sweepableGuard :: Text
sweepableGuard =
  let spent :: Text
spent = Text -> Text
attemptsSpentSQL Text
""
   in [text|
        NOT suspended AND cancel_requested_at IS NULL
        AND ${spent}
        AND (not_visible_until IS NULL OR not_visible_until <= NOW())
      |]

-- | Move a job to the DLQ in one statement. Copy the carried job columns
-- ('Arbiter.Core.Sql.Jobs.dlqCarriedCols') and the failure message.
moveToDLQSQL :: DLQMove -> SchemaName -> TableName -> Int64 -> Int64 -> Text -> Query Int64
moveToDLQSQL :: DLQMove -> Text -> Text -> Int64 -> Int64 -> Text -> Query Int64
moveToDLQSQL DLQMove
move Text
schemaName Text
tableName Int64
jobId Int64
cseq Text
errorMsg =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      exhausted :: Text
exhausted = Bool -> Text -> Text
forall m. Monoid m => Bool -> m -> m
mwhen (DLQMove
move DLQMove -> DLQMove -> Bool
forall a. Eq a => a -> a -> Bool
== DLQMove
MoveIfExhausted) [text|AND ${sweepableGuard}|]
   in [sql|
        WITH deleted_job AS (
          DELETE FROM ${tbl}
          WHERE id = #{jobId :: CInt8} AND claim_seq = #{cseq :: CInt8} ${exhausted}
          RETURNING *
        ),
        inserted_dlq AS (
          INSERT INTO ${dlqTbl} (job_id, ${dlqCarriedCols}, last_error)
          SELECT id, ${dlqCarriedCols}, #{errorMsg :: CText}
          FROM deleted_job
        )
        SELECT count(*) AS @{count :: CInt8} FROM deleted_job
      |]

-- | Up to @limit@ visible, unsuspended, uncancelled jobs with no attempts left.
selectExhaustedJobsSQL :: SchemaName -> TableName -> Int -> Query (Int64, Int64, Maybe Int64, Bool)
selectExhaustedJobsSQL :: Text -> Text -> Int -> Query (Int64, Int64, Maybe Int64, Bool)
selectExhaustedJobsSQL Text
schemaName Text
tableName Int
limit =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      lim :: Text
lim = Int -> Text
forall a. Integral a => a -> Text
intLiteral Int
limit
   in [sql|
        SELECT @{id :: CInt8}, @{claim_seq :: CInt8}, @{parent_id :: Maybe CInt8},
               (parent_state IS NOT NULL) AS @{is_rollup :: CBool}
        FROM ${tbl}
        WHERE ${sweepableGuard}
        ORDER BY id ASC
        LIMIT ${lim}
      |]

-- | Retry a DLQ job and its complete DLQ tree in one statement. Any member
-- identifies the tree. Restore the root, all descendants in the DLQ, and their
-- finalizers. Keep a finalizer suspended when it has children in the retry or the
-- main queue. Make it ready when it has no children. Re-suspend a main-queue
-- rollup parent that gains restored children, unless it is under a live lease. Refuse a
-- root whose parent is absent from the main queue. Remove the deduplication key during
-- the retry. An @edit@ replaces its columns on the target row only. Returns the
-- target's restored row.
retryFromDLQSQL :: SchemaName -> TableName -> Int64 -> Maybe RowEdit -> Query (JobRead (Stored payload))
retryFromDLQSQL :: forall payload.
Text
-> Text
-> Int64
-> Maybe RowEdit
-> Query (JobRead (Stored payload))
retryFromDLQSQL Text
schemaName Text
tableName Int64
dlqId Maybe RowEdit
edit =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      joinedEdit :: Query ()
joinedEdit = Maybe RowEdit -> Query ()
editJoin Maybe RowEdit
edit
      vals :: Text
vals = [Text] -> (Text -> Text) -> Text
requeuedValsEditing ((RowEdit -> [Text]) -> Maybe RowEdit -> [Text]
forall m a. Monoid m => (a -> m) -> Maybe a -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap RowEdit -> [Text]
editColumns Maybe RowEdit
edit) Text -> Text
targetEdited
      walk :: Query ()
walk = Text -> Text -> Int64 -> Query ()
dlqRetryTreeCte Text
dlqTbl Text
tbl Int64
dlqId
   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|
        ${walk},
        -- The root's parent is NULL or exists in the main queue
        can_retry AS (
          SELECT EXISTS (
            SELECT 1
            FROM root_job_id root
            JOIN ${dlqTbl} dead ON dead.job_id = root.job_id
            WHERE dead.parent_id IS NULL
               OR EXISTS (SELECT 1 FROM ${tbl} WHERE id = dead.parent_id)
          ) AS val
        ),
        -- Delete all tree members from DLQ (guarded by can_retry)
        deleted AS (
          DELETE FROM ${dlqTbl}
          WHERE id IN (SELECT dlq_id FROM tree)
            AND (SELECT val FROM can_retry)
          RETURNING job_id, claim_seq, ${requeuedCols}
        ),
        -- Re-insert into the main queue. A rollup finalizer is suspended when
        -- it has children in this retry batch or in the main queue.
        inserted AS (
          INSERT INTO ${tbl} (id, attempts, claim_seq, suspended, ${requeuedCols})
          SELECT dead.job_id, 0, dead.claim_seq + 1,
                 CASE WHEN dead.parent_state IS NOT NULL
                   THEN EXISTS (SELECT 1 FROM deleted child WHERE child.parent_id = dead.job_id)
                     OR EXISTS (SELECT 1 FROM ${tbl} WHERE parent_id = dead.job_id)
                   ELSE FALSE
                 END,
                 ${vals}
          FROM deleted dead ${joinedEdit}
          RETURNING *
        ),
        -- Re-suspend any rollup parents already in the main queue that just
        -- got fresh children re-inserted under them.
        parent_resuspend AS (
          UPDATE ${tbl}
          SET suspended = TRUE, claimed_by = NULL, updated_at = NOW()
          WHERE parent_state IS NOT NULL
            AND id IN (SELECT DISTINCT parent_id FROM inserted WHERE parent_id IS NOT NULL)
            AND NOT suspended
            AND NOT (claimed_by IS NOT NULL AND not_visible_until IS NOT NULL AND not_visible_until > NOW())
        )
        SELECT ${jobReadColumns} FROM inserted WHERE id = (SELECT job_id FROM target)
      |]

-- | Recursive CTEs binding @root_job_id@ to the root of the DLQ tree that holds
-- @dlqId@, and @tree@ to that tree's members.
dlqRetryTreeCte :: Text -> Text -> Int64 -> Query ()
dlqRetryTreeCte :: Text -> Text -> Int64 -> Query ()
dlqRetryTreeCte Text
dlqTbl Text
tbl Int64
dlqId =
  [sql|
    WITH RECURSIVE
    target AS (
      SELECT * FROM ${dlqTbl} WHERE id = #{dlqId :: CInt8}
    ),
    -- Walk up through DLQ ancestors to find the root of the tree.
    -- Stops when parent_id IS NULL, parent is in main queue, or
    -- parent is not found in DLQ (orphaned).
    ancestors AS (
      SELECT dead.job_id, dead.parent_id, 0 AS depth
      FROM ${dlqTbl} dead
      WHERE dead.job_id = (SELECT parent_id FROM target)
        AND (SELECT parent_id FROM target) IS NOT NULL
        AND NOT EXISTS (SELECT 1 FROM ${tbl} WHERE id = (SELECT parent_id FROM target))
      UNION ALL
      SELECT dead.job_id, dead.parent_id, ancestor.depth + 1
      FROM ${dlqTbl} dead
      JOIN ancestors ancestor ON dead.job_id = ancestor.parent_id
      WHERE ancestor.parent_id IS NOT NULL
        AND NOT EXISTS (SELECT 1 FROM ${tbl} WHERE id = ancestor.parent_id)
    ),
    -- Root is the topmost DLQ ancestor, or the target itself
    root_job_id AS (
      SELECT COALESCE(
        (SELECT job_id FROM ancestors ORDER BY depth DESC LIMIT 1),
        (SELECT job_id FROM target)
      ) AS job_id
    ),
    -- Walk down from root to collect all DLQ tree members
    tree AS (
      SELECT dead.id AS dlq_id, dead.job_id
      FROM ${dlqTbl} dead
      WHERE dead.job_id = (SELECT job_id FROM root_job_id)
      UNION ALL
      SELECT dead.id AS dlq_id, dead.job_id
      FROM ${dlqTbl} dead
      JOIN tree member ON dead.parent_id = member.job_id
    )
  |]

-- | The main-queue parent of the tree a DLQ retry restores.
dlqRetryParentSQL :: SchemaName -> TableName -> Int64 -> Query Int64
dlqRetryParentSQL :: Text -> Text -> Int64 -> Query Int64
dlqRetryParentSQL Text
schemaName Text
tableName Int64
dlqId = Text -> Text -> Int64 -> Text -> Query Int64
dlqRetryParent Text
schemaName Text
tableName Int64
dlqId Text
""

-- | Lock the main-queue parent of the tree a DLQ retry restores.
lockDLQRetryParentSQL :: SchemaName -> TableName -> Int64 -> Query Int64
lockDLQRetryParentSQL :: Text -> Text -> Int64 -> Query Int64
lockDLQRetryParentSQL Text
schemaName Text
tableName Int64
dlqId = Text -> Text -> Int64 -> Text -> Query Int64
dlqRetryParent Text
schemaName Text
tableName Int64
dlqId Text
"FOR UPDATE"

dlqRetryParent :: SchemaName -> TableName -> Int64 -> Text -> Query Int64
dlqRetryParent :: Text -> Text -> Int64 -> Text -> Query Int64
dlqRetryParent Text
schemaName Text
tableName Int64
dlqId Text
lockClause =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      walk :: Query ()
walk = Text -> Text -> Int64 -> Query ()
dlqRetryTreeCte Text
dlqTbl Text
tbl Int64
dlqId
   in [sql|
        ${walk}
        SELECT @{id :: CInt8} FROM ${tbl}
        WHERE id = (SELECT dead.parent_id FROM ${dlqTbl} dead WHERE dead.job_id = (SELECT job_id FROM root_job_id))
        ${lockClause}
      |]

-- | Create and lock, in key order, the group summaries of the tree a DLQ retry restores
-- and of its parent.
lockDLQRetryGroupsSQL :: SchemaName -> TableName -> Int64 -> Query ()
lockDLQRetryGroupsSQL :: Text -> Text -> Int64 -> Query ()
lockDLQRetryGroupsSQL Text
schemaName Text
tableName Int64
dlqId =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      groupsTbl :: Text
groupsTbl = Text -> Text -> Text
jobQueueGroupsTable Text
schemaName Text
tableName
      walk :: Query ()
walk = Text -> Text -> Int64 -> Query ()
dlqRetryTreeCte Text
dlqTbl Text
tbl Int64
dlqId
   in [sql|
        ${walk}
        INSERT INTO ${groupsTbl} (group_key)
        SELECT DISTINCT group_key FROM (
          SELECT dead.group_key FROM ${dlqTbl} dead WHERE dead.job_id IN (SELECT job_id FROM tree)
          UNION
          SELECT parent.group_key FROM ${tbl} parent
          JOIN ${dlqTbl} dead ON parent.id = dead.parent_id
          WHERE dead.job_id = (SELECT job_id FROM root_job_id)
        ) keys
        WHERE group_key IS NOT NULL
        ORDER BY group_key
        ${lockGroupKeyOnConflictSQL}
      |]

-- | An edited column, taken from the edit on the target row and kept on the others.
targetEdited :: Text -> Text
targetEdited :: Text -> Text
targetEdited Text
column =
  [text|CASE WHEN dead.job_id = (SELECT job_id FROM target) THEN edit.${column} ELSE dead.${column} END|]

-- | Whether a DLQ job with the given id exists.
dlqJobExistsSQL :: SchemaName -> TableName -> Int64 -> Query Bool
dlqJobExistsSQL :: Text -> Text -> Int64 -> Query Bool
dlqJobExistsSQL Text
schemaName Text
tableName Int64
dlqId =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
   in [sql|SELECT EXISTS (SELECT 1 FROM ${dlqTbl} WHERE id = #{dlqId :: CInt8}) AS @{result :: CBool}|]

-- | 'moveToDLQSQL' over @unnest@ed @(id, claim_seq, error_msg)@ arrays. Locks rows in
-- descending id order. Returns the ids moved.
moveToDLQBatchSQL :: SchemaName -> TableName -> [Int64] -> [Int64] -> [Text] -> Query Int64
moveToDLQBatchSQL :: Text -> Text -> [Int64] -> [Int64] -> [Text] -> Query Int64
moveToDLQBatchSQL Text
schemaName Text
tableName [Int64]
ids [Int64]
cseqs [Text]
errs =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      locked :: Query ()
locked = Text -> [Int64] -> Query ()
lockedByIdsCte Text
tbl [Int64]
ids
   in [sql|
        WITH input_jobs AS (
          SELECT unnest(#{ids :: [CInt8]}::bigint[]) AS id,
                 unnest(#{cseqs :: [CInt8]}::bigint[]) AS expected_claim_seq,
                 unnest(#{errs :: [CText]}::text[]) AS error_msg
        ),
        ${locked},
        deleted_jobs AS (
          DELETE FROM ${tbl} job
          USING input_jobs input_job
          WHERE job.id = input_job.id AND job.claim_seq = input_job.expected_claim_seq
            AND job.id IN (SELECT id FROM locked)
          RETURNING job.*, input_job.error_msg AS new_error
        ),
        inserted_dlq AS (
          INSERT INTO ${dlqTbl} (job_id, failed_at, ${dlqCarriedCols}, last_error)
          SELECT id, NOW(), ${dlqCarriedCols}, new_error
          FROM deleted_jobs
        )
        SELECT id AS @{result :: CInt8} FROM deleted_jobs
      |]

-- | Delete DLQ jobs by id, returning each one's id and parent id.
deleteDLQJobsBatchSQL :: SchemaName -> TableName -> [Int64] -> Query (Int64, Maybe Int64)
deleteDLQJobsBatchSQL :: Text -> Text -> [Int64] -> Query (Int64, Maybe Int64)
deleteDLQJobsBatchSQL Text
schemaName Text
tableName [Int64]
dlqIds =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
   in [sql|
        DELETE FROM ${dlqTbl} WHERE id = ANY(#{dlqIds :: [CInt8]})
        RETURNING @{id :: CInt8}, @{parent_id :: Maybe CInt8}
      |]

-- | Up to @limit@ DLQ ids, oldest first, of rows that failed more than @retention@
-- seconds ago.
selectExpiredDLQSQL :: SchemaName -> TableName -> Int64 -> Int -> Query Int64
selectExpiredDLQSQL :: Text -> Text -> Int64 -> Int -> Query Int64
selectExpiredDLQSQL Text
schemaName Text
tableName Int64
retention Int
limit =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      lim :: Text
lim = Int -> Text
forall a. Integral a => a -> Text
intLiteral Int
limit
   in [sql|
        SELECT @{id :: CInt8} FROM ${dlqTbl}
        WHERE failed_at < NOW() - (#{retention :: CInt8} * interval '1 second')
        ORDER BY failed_at
        LIMIT ${lim}
      |]

-- | The parents of the given DLQ rows.
dlqParentIdsSQL :: SchemaName -> TableName -> [Int64] -> Query (Maybe Int64)
dlqParentIdsSQL :: Text -> Text -> [Int64] -> Query (Maybe Int64)
dlqParentIdsSQL Text
schemaName Text
tableName [Int64]
dlqIds =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
   in [sql|SELECT @{parent_id :: Maybe CInt8} FROM ${dlqTbl} WHERE id = ANY(#{dlqIds :: [CInt8]})|]

-- | Move every descendant of a rollup parent to the DLQ alongside it.
cascadeChildrenToDLQSQL :: SchemaName -> TableName -> Int64 -> Text -> Query Int64
cascadeChildrenToDLQSQL :: Text -> Text -> Int64 -> Text -> Query Int64
cascadeChildrenToDLQSQL Text
schemaName Text
tableName Int64
parentId Text
errorMsg =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      cte :: Query ()
cte = Text -> Int64 -> Query ()
childDescendantsCte Text
tbl Int64
parentId
   in [sql|
        ${cte},
        deleted AS (
          DELETE FROM ${tbl}
          WHERE id IN (SELECT id FROM descendants)
          RETURNING id, ${dlqCarriedCols}
        ),
        inserted_dlq AS (
          INSERT INTO ${dlqTbl} (job_id, ${dlqCarriedCols}, last_error)
          SELECT id, ${dlqCarriedCols}, #{errorMsg :: CText}
          FROM deleted
        )
        SELECT count(*) AS @{count :: CInt8} FROM deleted
      |]

-- | DLQ child count per parent, over a set of job ids.
countDLQChildrenBatchSQL :: SchemaName -> TableName -> [Int64] -> Query (Int64, Int64)
countDLQChildrenBatchSQL :: Text -> Text -> [Int64] -> Query (Int64, Int64)
countDLQChildrenBatchSQL Text
schemaName Text
tableName [Int64]
jobIds =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
   in [sql|
        SELECT @{parent_id :: CInt8}, COUNT(*) AS @{count :: CInt8}
        FROM ${dlqTbl}
        WHERE parent_id = ANY(#{jobIds :: [CInt8]})
        GROUP BY parent_id
      |]

-- | List DLQ jobs under a dynamic WHERE and an @orderBy@ from 'Arbiter.Core.Sql.Jobs.buildDLQOrderBy'.
listDLQFilteredSQL
  :: SchemaName -> TableName -> Query () -> Text -> Int64 -> Int64 -> Query (Int64, UTCTime, JobRead (Stored payload))
listDLQFilteredSQL :: forall payload.
Text
-> Text
-> Query ()
-> Text
-> Int64
-> Int64
-> Query (Int64, UTCTime, JobRead (Stored payload))
listDLQFilteredSQL Text
schemaName Text
tableName Query ()
whereFrag Text
orderBy Int64
limit Int64
offset =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
   in RowCodec (Int64, UTCTime, JobRead (Stored payload))
-> Query () -> Query (Int64, UTCTime, JobRead (Stored payload))
forall a. RowCodec a -> Query () -> Query a
rows
        (Text -> RowCodec (Int64, UTCTime, JobRead (Stored payload))
forall payload.
Text -> RowCodec (Int64, UTCTime, JobRead (Stored payload))
dlqRowCodec Text
tableName)
        [sql|
          SELECT ${allDLQColumns}
          FROM ${dlqTbl}
          ${whereFrag}
          ORDER BY ${orderBy}
          LIMIT #{limit :: CInt8} OFFSET #{offset :: CInt8}
        |]

-- | The ids of the DLQ jobs under a dynamic WHERE, oldest first.
dlqIdsFilteredSQL :: SchemaName -> TableName -> Query () -> Query Int64
dlqIdsFilteredSQL :: Text -> Text -> Query () -> Query Int64
dlqIdsFilteredSQL Text
schemaName Text
tableName Query ()
whereFrag =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
   in [sql|SELECT @{id :: CInt8} FROM ${dlqTbl} ${whereFrag} ORDER BY id|]

-- | Count DLQ jobs under a dynamic WHERE.
countDLQFilteredSQL :: SchemaName -> TableName -> Query () -> Query Int64
countDLQFilteredSQL :: Text -> Text -> Query () -> Query Int64
countDLQFilteredSQL Text
schemaName Text
tableName Query ()
whereFrag =
  let dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
   in Text -> Query () -> Query Int64
countWhereSQL Text
dlqTbl Query ()
whereFrag