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

-- | Internal to the arbiter packages. Not covered by the PVP.
--
-- Schema for the per-queue group summary table and the statement-level triggers
-- that maintain it. The summary carries each group's head, counts and visibility
-- deadlines. The claim ranks groups from it.
module Arbiter.Core.Job.Schema.Groups
  ( -- * Grouped Job Indexes
    createJobQueueGroupKeyIndexSQL
  , createJobQueueGroupRetriedIndexSQL
  , createJobQueueGroupedDueIndexSQL
  , createJobQueueGroupInFlightIndexSQL

    -- * Groups table SQL
  , createGroupsTableSQL
  , migrateGroupsReadyRankingSQL
  , createGroupsEmptiedIndexSQL

    -- * Summary column definitions
  , groupAggregates
  , inFlightPredicate

    -- * Groups trigger SQL
  , createGroupsTriggerFunctionsSQL
  , createGroupsTriggersSQL
  ) where

import Data.Text (Text)
import Data.Text qualified as T
import NeatInterpolation (text)

import Arbiter.Core.Job.Schema
  ( SchemaName
  , TableName
  , createMaintenanceTriggersSQL
  , indexSQL
  , jobQueueGroupsTable
  , jobQueueTable
  , maintenanceFunctionNames
  , qualifiedName
  )
import Arbiter.Core.SqlLiterals (attemptsLeftSQL, dollarQuote, lockGroupKeyOnConflictSQL, quoteIdentifier)

-- | Partial index over @(group_key, priority, id)@, read by the claim's LATERAL
-- subqueries and by the maintenance triggers recomputing a group's minima.
createJobQueueGroupKeyIndexSQL :: SchemaName -> TableName -> Text
createJobQueueGroupKeyIndexSQL :: Text -> Text -> Text
createJobQueueGroupKeyIndexSQL Text
schemaName Text
tableName =
  Text -> Text -> Text -> Maybe Text -> Text
indexSQL
    (Text
"idx_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_group_key")
    (Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName)
    Text
"group_key, priority ASC, id ASC"
    (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"group_key IS NOT NULL")

-- | Partial index over @(group_key, attempts DESC, priority, id)@ for retried rows,
-- read by the claim's group head gate. Retried rows rank ahead of the rest. The
-- gate merges this run with the @(group_key, priority, id)@ scan.
createJobQueueGroupRetriedIndexSQL :: SchemaName -> TableName -> Text
createJobQueueGroupRetriedIndexSQL :: Text -> Text -> Text
createJobQueueGroupRetriedIndexSQL Text
schemaName Text
tableName =
  Text -> Text -> Text -> Maybe Text -> Text
indexSQL
    (Text
"idx_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_group_retried")
    (Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName)
    Text
"group_key, attempts DESC, priority ASC, id ASC"
    (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"group_key IS NOT NULL AND attempts > 0")

-- | Parked grouped jobs (scheduled, backoff or leased) by visibility deadline. Group
-- maintenance uses this index to replace @next_due@ with one point lookup.
createJobQueueGroupedDueIndexSQL :: SchemaName -> TableName -> Text
createJobQueueGroupedDueIndexSQL :: Text -> Text -> Text
createJobQueueGroupedDueIndexSQL Text
schemaName Text
tableName =
  Text -> Text -> Text -> Maybe Text -> Text
indexSQL
    (Text
"idx_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_grouped_due")
    (Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName)
    Text
"group_key, not_visible_until ASC"
    (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"group_key IS NOT NULL AND not_visible_until IS NOT NULL AND NOT suspended")

-- | Possible in-flight grouped jobs by descending lease deadline. The query
-- applies the time-dependent part of the predicate at runtime.
createJobQueueGroupInFlightIndexSQL :: SchemaName -> TableName -> Text
createJobQueueGroupInFlightIndexSQL :: Text -> Text -> Text
createJobQueueGroupInFlightIndexSQL Text
schemaName Text
tableName =
  Text -> Text -> Text -> Maybe Text -> Text
indexSQL
    (Text
"idx_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_group_in_flight")
    (Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName)
    Text
"group_key, not_visible_until DESC NULLS LAST"
    ( Text -> Maybe Text
forall a. a -> Maybe a
Just
        Text
"group_key IS NOT NULL AND not_visible_until IS NOT NULL AND NOT suspended AND (attempts > 0 OR throttled_until IS NOT NULL)"
    )

-- | Create a queue's groups table, one summary row per @group_key@ carrying the group's
-- precomputed minima, counts and @in_flight_until@. Maintained by the triggers
-- 'createGroupsTriggersSQL' installs.
createGroupsTableSQL :: SchemaName -> TableName -> Text
createGroupsTableSQL :: Text -> Text -> Text
createGroupsTableSQL Text
schemaName Text
tableName =
  let groupsTbl :: Text
groupsTbl = Text -> Text -> Text
jobQueueGroupsTable Text
schemaName Text
tableName
   in [Text] -> Text
T.unlines
        [ Text
"CREATE TABLE IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
groupsTbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ("
        , Text
"  group_key TEXT PRIMARY KEY,"
        , Text
"  min_priority INT NOT NULL DEFAULT 0,"
        , Text
"  min_id BIGINT NOT NULL DEFAULT 0,"
        , Text
"  job_count INT NOT NULL DEFAULT 0,"
        , Text
"  in_flight_until TIMESTAMPTZ DEFAULT NULL"
        , Text
");"
        ]

-- | Add @ready_count@ and @next_due@ to the groups summary, make the ranking
-- index partial on ready rows, and add the @next_due@ due-finder.
migrateGroupsReadyRankingSQL :: SchemaName -> TableName -> Text
migrateGroupsReadyRankingSQL :: Text -> Text -> Text
migrateGroupsReadyRankingSQL Text
schemaName Text
tableName =
  let groupsTbl :: Text
groupsTbl = Text -> Text -> Text
jobQueueGroupsTable Text
schemaName Text
tableName
      tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      qidx :: Text -> Text
qidx Text
suffix = Text -> Text
quoteIdentifier (Text
"idx_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
suffix)
   in [Text] -> Text
T.unlines
        [ Text
"ALTER TABLE " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
groupsTbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ADD COLUMN IF NOT EXISTS ready_count INT NOT NULL DEFAULT 0;"
        , Text
"ALTER TABLE " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
groupsTbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ADD COLUMN IF NOT EXISTS next_due TIMESTAMPTZ;"
        , Text
"UPDATE " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
groupsTbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" g SET"
        , Text
"  min_priority = sub.mp, min_id = sub.mi, ready_count = COALESCE(sub.rc, 0), next_due = sub.nd"
        , Text
"FROM ("
        , Text
"  SELECT group_key,"
        , Text
"    MIN(priority) AS mp,"
        , Text
"    MIN(id) AS mi,"
        , Text
"    COUNT(*) FILTER (WHERE not_visible_until IS NULL AND NOT suspended) AS rc,"
        , Text
"    MIN(not_visible_until) FILTER (WHERE not_visible_until IS NOT NULL AND NOT suspended) AS nd"
        , Text
"  FROM " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" WHERE group_key IS NOT NULL GROUP BY group_key"
        , Text
") sub WHERE g.group_key = sub.group_key;"
        , Text
"DROP INDEX IF EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
qualifiedName Text
schemaName (Text
"idx_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_groups_ranking") Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"
        , Text
"CREATE INDEX IF NOT EXISTS "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
qidx Text
"_groups_ranking"
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
groupsTbl
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (min_priority ASC, min_id ASC) WHERE ready_count > 0 AND in_flight_until IS NULL;"
        , Text
"CREATE INDEX IF NOT EXISTS "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
qidx Text
"_groups_next_due"
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
groupsTbl
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (next_due ASC) WHERE next_due IS NOT NULL;"
        ]

-- | Index over the summary rows the maintenance triggers emptied in place.
createGroupsEmptiedIndexSQL :: SchemaName -> TableName -> Text
createGroupsEmptiedIndexSQL :: Text -> Text -> Text
createGroupsEmptiedIndexSQL Text
schemaName Text
tableName =
  Text -> Text -> Text -> Maybe Text -> Text
indexSQL
    (Text
"idx_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_groups_emptied")
    (Text -> Text -> Text
jobQueueGroupsTable Text
schemaName Text
tableName)
    Text
"group_key"
    (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"job_count = 0")

-- | The @ON CONFLICT@ merge of newly grouped rows into an existing group summary.
-- The head is whichever side ranks first as a whole @(min_priority, min_id)@ pair.
groupsMergeSet :: Text -> Text
groupsMergeSet :: Text -> Text
groupsMergeSet Text
groupsTbl =
  let takesHead :: Text
takesHead =
        [text|${groupsTbl}.job_count = 0 OR (EXCLUDED.min_priority, EXCLUDED.min_id) < (${groupsTbl}.min_priority, ${groupsTbl}.min_id)|]
   in [text|
    min_priority = CASE WHEN ${takesHead} THEN EXCLUDED.min_priority
      ELSE ${groupsTbl}.min_priority END,
    min_id = CASE WHEN ${takesHead} THEN EXCLUDED.min_id
      ELSE ${groupsTbl}.min_id END,
    job_count = ${groupsTbl}.job_count + EXCLUDED.job_count,
    ready_count = ${groupsTbl}.ready_count + EXCLUDED.ready_count,
    next_due = LEAST(${groupsTbl}.next_due, EXCLUDED.next_due)
  |]

-- | Lock a queue's existing group summaries in @group_key@ order.
groupsLock :: Text -> Text -> Text
groupsLock :: Text -> Text -> Text
groupsLock Text
groupsTbl Text
keys =
  [text|
    PERFORM 1 FROM ${groupsTbl} g
    WHERE g.group_key IN (${keys})
    ORDER BY g.group_key FOR UPDATE;
  |]

-- | Create the missing group summaries and lock the existing ones, in one @group_key@
-- order pass. A row another transaction is creating is waited on in its turn.
groupsCreateLock :: Text -> Text -> Text
groupsCreateLock :: Text -> Text -> Text
groupsCreateLock Text
groupsTbl Text
keys =
  [text|
    INSERT INTO ${groupsTbl} (group_key)
    SELECT DISTINCT group_key FROM (${keys}) created ORDER BY group_key
    ${lockGroupKeyOnConflictSQL};
  |]

groupsInsertFunction :: Text -> Text -> Text
groupsInsertFunction :: Text -> Text -> Text
groupsInsertFunction Text
funcName Text
groupsTbl =
  let mergeSet :: Text
mergeSet = Text -> Text
groupsMergeSet Text
groupsTbl
      aggs :: Text
aggs = Text -> Text
groupAggregates Text
""
      lockRows :: Text
lockRows = Text -> Text -> Text
groupsCreateLock Text
groupsTbl Text
"SELECT group_key FROM new_table WHERE group_key IS NOT NULL"
   in [text|
    CREATE OR REPLACE FUNCTION ${funcName}()
    RETURNS TRIGGER AS ${dollarQuote}
    BEGIN
      IF NOT EXISTS (SELECT 1 FROM new_table WHERE group_key IS NOT NULL LIMIT 1) THEN
        RETURN NULL;
      END IF;

      ${lockRows}

      INSERT INTO ${groupsTbl} (group_key, min_priority, min_id, job_count, ready_count, next_due)
      SELECT group_key,
        ${aggs}
      FROM new_table
      WHERE group_key IS NOT NULL
      GROUP BY group_key
      ORDER BY group_key
      ON CONFLICT (group_key) DO UPDATE SET
        ${mergeSet},
        in_flight_until = CASE WHEN ${groupsTbl}.in_flight_until <= NOW()
          THEN NULL ELSE ${groupsTbl}.in_flight_until END;

      RETURN NULL;
    END;
    ${dollarQuote} LANGUAGE plpgsql;
  |]

-- | The group summary aggregates over job rows grouped by @group_key@. @min_priority@
-- and @min_id@ rank the group. @col@ prefixes each column.
groupAggregates :: Text -> Text
groupAggregates :: Text -> Text
groupAggregates Text
col =
  let ready :: Text
ready = Text -> Text
readyPredicate Text
col
      scheduled :: Text
scheduled = Text -> Text
scheduledPredicate Text
col
   in [text|
    MIN(${col}priority) AS min_priority,
    (MIN(ARRAY[${col}priority::bigint, ${col}id]))[2] AS min_id,
    COUNT(*) AS job_count,
    COUNT(*) FILTER (WHERE ${ready}) AS ready_count,
    MIN(${col}not_visible_until) FILTER (WHERE ${scheduled}) AS next_due
  |]

-- | Whether a job counts toward its group's ready total. @col@ prefixes each column.
readyPredicate :: Text -> Text
readyPredicate :: Text -> Text
readyPredicate = Text -> Text -> Text
summaryPredicate Text
"IS NULL"

-- | Whether a job's deadline counts toward its group's next due time. @col@ prefixes
-- each column.
scheduledPredicate :: Text -> Text
scheduledPredicate :: Text -> Text
scheduledPredicate = Text -> Text -> Text
summaryPredicate Text
"IS NOT NULL"

-- | A live, unsuspended job with attempts left whose @not_visible_until@ passes @visibility@.
summaryPredicate :: Text -> Text -> Text
summaryPredicate :: Text -> Text -> Text
summaryPredicate Text
visibility Text
col =
  let attemptsLeft :: Text
attemptsLeft = Text -> Text
attemptsLeftSQL Text
col
   in [text|${col}not_visible_until ${visibility} AND NOT ${col}suspended AND ${col}cancel_requested_at IS NULL AND ${attemptsLeft}|]

-- | Whether a job still holds its group's in-flight slot. @col@ prefixes each column.
inFlightPredicate :: Text -> Text
inFlightPredicate :: Text -> Text
inFlightPredicate Text
col =
  [text|${col}not_visible_until > NOW() AND NOT ${col}suspended AND (${col}attempts > 0 OR ${col}throttled_until > NOW())|]

-- | Whether an updated row changes a value some group summary reads, its
-- @group_key@ aside.
summaryValueChanged :: Text
summaryValueChanged :: Text
summaryValueChanged =
  Text
"n.priority IS DISTINCT FROM o.priority"
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" OR n.not_visible_until IS DISTINCT FROM o.not_visible_until"
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" OR n.suspended IS DISTINCT FROM o.suspended"
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" OR n.attempts IS DISTINCT FROM o.attempts"
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" OR n.max_attempts IS DISTINCT FROM o.max_attempts"
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" OR n.cancel_requested_at IS DISTINCT FROM o.cancel_requested_at"
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" OR n.throttled_until IS DISTINCT FROM o.throttled_until"

-- | Apply a group's count deltas and replace all four extrema with indexed point
-- lookups. @deltaRows@ supplies @group_key@, @count_delta@ and @ready_delta@. Each
-- lookup reads one index entry and stops. A group whose summary comes out unchanged
-- keeps its row.
groupsSnapshotUpdate :: Text -> Text -> Text -> Text
groupsSnapshotUpdate :: Text -> Text -> Text -> Text
groupsSnapshotUpdate Text
groupsTbl Text
tbl Text
deltaRows =
  let scheduled :: Text
scheduled = Text -> Text
scheduledPredicate Text
"q."
      inFlight :: Text
inFlight = Text -> Text
inFlightPredicate Text
"q."
      emptied :: Text
emptied = Text
"hp.priority IS NULL"
   in [text|
    WITH t AS (
      SELECT g0.group_key,
        CASE WHEN ${emptied} THEN 0 ELSE GREATEST(0, g0.job_count + d.count_delta) END AS job_count,
        CASE WHEN ${emptied} THEN 0 ELSE GREATEST(0, g0.ready_count + d.ready_delta) END AS ready_count,
        COALESCE(hp.priority, 0) AS min_priority,
        COALESCE(hp.id, 0) AS min_id,
        nd.not_visible_until AS next_due,
        fi.not_visible_until AS in_flight_until
      FROM (
        ${deltaRows}
      ) d
      JOIN ${groupsTbl} g0 ON g0.group_key = d.group_key
      LEFT JOIN LATERAL (
        SELECT q.priority, q.id FROM ${tbl} q
        WHERE q.group_key = d.group_key
        ORDER BY q.priority ASC, q.id ASC LIMIT 1
      ) hp ON TRUE
      LEFT JOIN LATERAL (
        SELECT q.not_visible_until FROM ${tbl} q
        WHERE q.group_key = d.group_key AND ${scheduled}
        ORDER BY q.not_visible_until ASC LIMIT 1
      ) nd ON TRUE
      LEFT JOIN LATERAL (
        SELECT q.not_visible_until FROM ${tbl} q
        WHERE q.group_key = d.group_key AND ${inFlight}
        ORDER BY q.not_visible_until DESC NULLS LAST LIMIT 1
      ) fi ON TRUE
    )
    UPDATE ${groupsTbl} g
    SET job_count = t.job_count,
      ready_count = t.ready_count,
      min_priority = t.min_priority,
      min_id = t.min_id,
      next_due = t.next_due,
      in_flight_until = t.in_flight_until
    FROM t
    WHERE g.group_key = t.group_key
      AND (g.job_count, g.ready_count, g.min_priority, g.min_id, g.next_due, g.in_flight_until)
          IS DISTINCT FROM (t.job_count, t.ready_count, t.min_priority, t.min_id, t.next_due, t.in_flight_until);
  |]

-- | Removed rows leave their group summary.
groupsDeleteFunction :: Text -> Text -> Text -> Text
groupsDeleteFunction :: Text -> Text -> Text -> Text
groupsDeleteFunction Text
funcName Text
groupsTbl Text
tbl =
  let ready :: Text
ready = Text -> Text
readyPredicate Text
""
      lockRows :: Text
lockRows = Text -> Text -> Text
groupsLock Text
groupsTbl Text
"SELECT group_key FROM old_table WHERE group_key IS NOT NULL"
      removeUpdate :: Text
removeUpdate =
        Text -> Text -> Text -> Text
groupsSnapshotUpdate
          Text
groupsTbl
          Text
tbl
          [text|
            SELECT group_key, (-COUNT(*))::int AS count_delta,
              (-COUNT(*) FILTER (WHERE ${ready}))::int AS ready_delta
            FROM old_table WHERE group_key IS NOT NULL GROUP BY group_key
          |]
   in [text|
    CREATE OR REPLACE FUNCTION ${funcName}()
    RETURNS TRIGGER AS ${dollarQuote}
    BEGIN
      IF NOT EXISTS (SELECT 1 FROM old_table WHERE group_key IS NOT NULL LIMIT 1) THEN
        RETURN NULL;
      END IF;

      ${lockRows}
      ${removeUpdate}
      RETURN NULL;
    END;
    ${dollarQuote} LANGUAGE plpgsql;
  |]

-- | Updated rows keep their group unless a dedup replace moved them. A moved row
-- leaves the old summary and merges into the new one. Counts come from the
-- transition tables. Everything else comes from the snapshot lookups.
groupsUpdateFunction :: Text -> Text -> Text -> Text
groupsUpdateFunction :: Text -> Text -> Text -> Text
groupsUpdateFunction Text
funcName Text
groupsTbl Text
tbl =
  let valueChanged :: Text
valueChanged = Text
summaryValueChanged
      changed :: Text
changed = Text
"n.group_key IS DISTINCT FROM o.group_key OR " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
valueChanged
      readyOld :: Text
readyOld = Text -> Text
readyPredicate Text
"o."
      readyNew :: Text
readyNew = Text -> Text
readyPredicate Text
"n."
      aggsN :: Text
aggsN = Text -> Text
groupAggregates Text
"n."
      mergeSet :: Text
mergeSet = Text -> Text
groupsMergeSet Text
groupsTbl
      lockRows :: Text
lockRows =
        Text -> Text -> Text
groupsCreateLock
          Text
groupsTbl
          [text|
            SELECT group_key
            FROM new_table
            WHERE group_key IS NOT NULL
            UNION
            SELECT group_key
            FROM old_table
            WHERE group_key IS NOT NULL
          |]
      departureUpdate :: Text
departureUpdate =
        Text -> Text -> Text -> Text
groupsSnapshotUpdate
          Text
groupsTbl
          Text
tbl
          [text|
            SELECT o.group_key, (-COUNT(*))::int AS count_delta,
              (-COUNT(*) FILTER (WHERE ${readyOld}))::int AS ready_delta
            FROM old_table o JOIN new_table n ON n.id = o.id
            WHERE o.group_key IS NOT NULL AND o.group_key IS DISTINCT FROM n.group_key
            GROUP BY o.group_key
          |]
      arrivalUpdate :: Text
arrivalUpdate =
        Text -> Text -> Text -> Text
groupsSnapshotUpdate
          Text
groupsTbl
          Text
tbl
          [text|
            SELECT n.group_key, 0 AS count_delta, 0 AS ready_delta
            FROM new_table n JOIN old_table o ON o.id = n.id
            WHERE n.group_key IS NOT NULL AND o.group_key IS DISTINCT FROM n.group_key
            GROUP BY n.group_key
          |]
      sameGroupUpdate :: Text
sameGroupUpdate =
        Text -> Text -> Text -> Text
groupsSnapshotUpdate
          Text
groupsTbl
          Text
tbl
          [text|
            SELECT n.group_key, 0 AS count_delta,
              SUM((${readyNew})::int - (${readyOld})::int)::int AS ready_delta
            FROM new_table n JOIN old_table o ON o.id = n.id
            WHERE n.group_key IS NOT NULL
              AND n.group_key IS NOT DISTINCT FROM o.group_key
              AND (${valueChanged})
            GROUP BY n.group_key
          |]
   in [text|
    CREATE OR REPLACE FUNCTION ${funcName}()
    RETURNS TRIGGER AS ${dollarQuote}
    BEGIN
      IF NOT EXISTS (SELECT 1 FROM new_table WHERE group_key IS NOT NULL LIMIT 1)
         AND NOT EXISTS (SELECT 1 FROM old_table WHERE group_key IS NOT NULL LIMIT 1) THEN
        RETURN NULL;
      END IF;

      IF NOT EXISTS (
        SELECT 1 FROM new_table n JOIN old_table o ON o.id = n.id
        WHERE (n.group_key IS NOT NULL OR o.group_key IS NOT NULL)
          AND (${changed})
        LIMIT 1
      ) THEN
        RETURN NULL;
      END IF;

      ${lockRows}
      IF EXISTS (
        SELECT 1 FROM new_table n JOIN old_table o ON o.id = n.id
        WHERE o.group_key IS DISTINCT FROM n.group_key LIMIT 1
      ) THEN
        ${departureUpdate}

        INSERT INTO ${groupsTbl} (group_key, min_priority, min_id, job_count, ready_count, next_due)
        SELECT n.group_key,
          ${aggsN}
        FROM new_table n JOIN old_table o ON o.id = n.id
        WHERE n.group_key IS NOT NULL AND o.group_key IS DISTINCT FROM n.group_key
        GROUP BY n.group_key ORDER BY n.group_key
        ON CONFLICT (group_key) DO UPDATE SET ${mergeSet};

        ${arrivalUpdate}
      END IF;
      ${sameGroupUpdate}
      RETURN NULL;
    END;
    ${dollarQuote} LANGUAGE plpgsql;
  |]

-- | Group maintenance with indexed extremum replacement. Transition tables
-- supply count deltas. No update or delete scans all rows in an affected group.
createGroupsTriggerFunctionsSQL :: SchemaName -> TableName -> Text
createGroupsTriggerFunctionsSQL :: Text -> Text -> Text
createGroupsTriggerFunctionsSQL Text
schemaName Text
tableName =
  let groupsTbl :: Text
groupsTbl = Text -> Text -> Text
jobQueueGroupsTable Text
schemaName Text
tableName
      tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      baseName :: Text
baseName = Text -> Text
groupsTriggerBase Text
tableName
      (Text
funcInsert, Text
funcDelete, Text
funcUpdate) = Text -> Text -> (Text, Text, Text)
maintenanceFunctionNames Text
schemaName Text
baseName
   in [Text] -> Text
T.unlines
        [ Text -> Text -> Text
groupsInsertFunction Text
funcInsert Text
groupsTbl
        , Text -> Text -> Text -> Text
groupsDeleteFunction Text
funcDelete Text
groupsTbl Text
tbl
        , Text -> Text -> Text -> Text
groupsUpdateFunction Text
funcUpdate Text
groupsTbl Text
tbl
        ]

-- | The three statement-level AFTER triggers calling a queue's groups maintenance
-- functions, each handed its affected rows through a transition table.
createGroupsTriggersSQL :: SchemaName -> TableName -> Text
createGroupsTriggersSQL :: Text -> Text -> Text
createGroupsTriggersSQL Text
schemaName Text
tableName =
  Text -> Text -> Text -> Text
createMaintenanceTriggersSQL Text
schemaName (Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName) (Text -> Text
groupsTriggerBase Text
tableName)

-- | Base name of a queue's groups trigger functions and triggers.
groupsTriggerBase :: TableName -> Text
groupsTriggerBase :: Text -> Text
groupsTriggerBase Text
tableName = Text
"maintain_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_groups"