{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE QuasiQuotes #-}

-- | SQL generation functions for job queue schemas. No database execution happens here.
module Arbiter.Core.Job.Schema
  ( -- * Name Types
    SchemaName
  , TableName

    -- * Schema Creation
  , createSchemaSQL
  , defaultSchemaName

    -- * Table Creation SQL
  , createJobQueueTableSQL
  , createJobQueueDLQTableSQL
  , createJobQueueArchiveTableSQL
  , queueTableNames
  , addTraceContextColumnSQL
  , addClaimSeqColumnSQL
  , addKindColumnSQL
  , setMaxAttemptsDefaultSQL

    -- * Index Creation SQL
  , createJobQueueUngroupedReadyRankingIndexSQL
  , createJobQueueUngroupedDueIndexSQL
  , migrateUngroupedReadySplitIndexesSQL
  , createDLQGroupKeyIndexSQL
  , createDLQFailedAtIndexSQL
  , createDLQParentIdIndexSQL
  , createArchiveCompletedAtIndexSQL
  , createArchiveExpiresAtIndexSQL
  , createArchiveJobIdIndexSQL
  , createArchiveParentIdIndexSQL
  , createArchiveGroupKeyIndexSQL
  , createDedupKeyIndexSQL
  , createParentIdIndexSQL

    -- * NOTIFY Trigger SQL
  , createNotifyFunctionSQL
  , createNotifyTriggerSQL
  , dropNotifyTriggerSQL
  , dropNotifyFunctionSQL

    -- * Event Streaming Trigger SQL
  , createEventStreamingFunctionSQL
  , createEventStreamingTriggersSQL
  , dropEventStreamingFunctionSQL
  , dropEventStreamingTriggersSQL

    -- * Notification Channel Helpers
  , notificationChannelForTable
  , eventStreamingChannel
  , pauseNotifyChannel
  , pauseNotifyChannelPrefix
  , cancelNotifyChannel
  , cancelNotifyChannelPrefix
  , cronRunNotifyChannel

    -- * Trigger / Function Name Helpers
  , notifyFunctionName
  , notifyTriggerName
  , eventStreamingFunctionName
  , eventStreamingTriggerName
  , eventStreamingDLQTriggerName
  , legacyEventStreamingTriggers
  , notifyObjectComment
  , notifyObjectCommentPrefix
  , notifyAdoptedObjectComment
  , eventStreamingObjectComment
  , eventStreamingObjectCommentPrefix
  , eventStreamingAdoptedObjectComment

    -- * Table Name Helpers
  , qualifiedTable
  , jobQueueTable
  , jobQueueDLQTable
  , jobQueueArchiveTable
  , jobQueueResultsTable
  , jobQueueGroupsTable

    -- * Results Table
  , createResultsTableSQL

    -- * Maintenance Trigger SQL
  , maintenanceFunctionNames
  , createMaintenanceTriggersSQL
  , statementTriggerSQL
  ) where

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

import Arbiter.Core.Job.Types (defaultMaxAttempts)
import Arbiter.Core.SqlLiterals (quoteIdentifier)

-- | PostgreSQL schema name, e.g. @"arbiter"@.
type SchemaName = Text

-- | Unqualified table name within a schema, e.g. @"email_jobs"@.
type TableName = Text

-- | The schema arbiter's tables live in by default.
defaultSchemaName :: SchemaName
defaultSchemaName :: Text
defaultSchemaName = Text
"arbiter"

-- | A table's own job-arrival NOTIFY channel: @\"email_jobs\"@ -> @\"email_jobs_created\"@.
notificationChannelForTable :: TableName -> Text
notificationChannelForTable :: Text -> Text
notificationChannelForTable Text
tableName = Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_created"

-- | Channel name used by the event streaming (SSE) system.
eventStreamingChannel :: Text
eventStreamingChannel :: Text
eventStreamingChannel = Text
"arbiter_job_events"

-- | Prefix for per-queue pause NOTIFY channels. The full channel name appends
-- the queue. SQL templates build the channel from @queue_name@ returned by a CTE.
pauseNotifyChannelPrefix :: SchemaName -> Text
pauseNotifyChannelPrefix :: Text -> Text
pauseNotifyChannelPrefix Text
schemaName = Text
"arbiter_pause_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_"

-- | Per-queue NOTIFY channel for pause/resume changes. Workers LISTEN on the
-- channel for their own queue.
pauseNotifyChannel :: SchemaName -> Text -> Text
pauseNotifyChannel :: Text -> Text -> Text
pauseNotifyChannel Text
schemaName Text
queueName =
  Int -> Text -> Text
T.take Int
63 (Text -> Text) -> Text -> Text
forall a b. (a -> b) -> a -> b
$ Text -> Text
pauseNotifyChannelPrefix Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
queueName

-- | Prefix for per-queue cancel NOTIFY channels. See 'cancelNotifyChannel'.
cancelNotifyChannelPrefix :: SchemaName -> Text
cancelNotifyChannelPrefix :: Text -> Text
cancelNotifyChannelPrefix Text
schemaName = Text
"arbiter_cancel_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_"

-- | Per-queue NOTIFY channel for force-cancel signals. The payload identifies
-- the target worker and job. Only the matching worker reacts.
cancelNotifyChannel :: SchemaName -> Text -> Text
cancelNotifyChannel :: Text -> Text -> Text
cancelNotifyChannel Text
schemaName Text
queueName =
  Int -> Text -> Text
T.take Int
63 (Text -> Text) -> Text -> Text
forall a b. (a -> b) -> a -> b
$ Text -> Text
cancelNotifyChannelPrefix Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
queueName

-- | Per-schema NOTIFY channel for manual cron run-now requests.
cronRunNotifyChannel :: SchemaName -> Text
cronRunNotifyChannel :: Text -> Text
cronRunNotifyChannel Text
schemaName = Int -> Text -> Text
T.take Int
63 (Text -> Text) -> Text -> Text
forall a b. (a -> b) -> a -> b
$ Text
"arbiter_cron_run_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
schemaName

-- | Per-table NOTIFY trigger function name.
notifyFunctionName :: TableName -> Text
notifyFunctionName :: Text -> Text
notifyFunctionName Text
tableName = Text
"notify_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_created"

-- | Per-table NOTIFY trigger name.
notifyTriggerName :: TableName -> Text
notifyTriggerName :: Text -> Text
notifyTriggerName Text
tableName = Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_notify_trigger"

-- | Shared event streaming trigger function name (one per schema).
eventStreamingFunctionName :: Text
eventStreamingFunctionName :: Text
eventStreamingFunctionName = Text
"notify_job_event"

-- | Per-table event streaming trigger name.
eventStreamingTriggerName :: TableName -> Text
eventStreamingTriggerName :: Text -> Text
eventStreamingTriggerName Text
tableName = Text
"notify_job_event_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName

-- | Per-table DLQ event streaming trigger name.
eventStreamingDLQTriggerName :: TableName -> Text
eventStreamingDLQTriggerName :: Text -> Text
eventStreamingDLQTriggerName Text
tableName = Text
"notify_job_event_" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_dlq"

-- | Event-streaming trigger names arbiter generated before the per-queue names, each
-- paired with whether it sits on the DLQ table.
legacyEventStreamingTriggers :: [(Text, Bool)]
legacyEventStreamingTriggers :: [(Text, Bool)]
legacyEventStreamingTriggers =
  [ (Text
"notify_job_insert", Bool
False)
  , (Text
"notify_job_update", Bool
False)
  , (Text
"notify_job_delete", Bool
False)
  , (Text
"notify_dlq_insert", Bool
True)
  ]

-- | Ownership marker stamped on every notify function and trigger arbiter installs.
-- Sweeps match 'notifyObjectCommentPrefix'. A trigger is current when its comment
-- equals this exact value. Bump the version whenever 'createNotifyTriggerSQL' changes.
notifyObjectComment :: Text
notifyObjectComment :: Text
notifyObjectComment = Text
notifyObjectCommentPrefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"v1"

-- | The marker prefix identifying a notify object as arbiter's, across versions.
notifyObjectCommentPrefix :: Text
notifyObjectCommentPrefix :: Text
notifyObjectCommentPrefix = Text
"arbiter:notify:"

-- | Marker stamped on notify objects installed before arbiter marked them. Sweeps
-- match it. An adopted trigger is rebuilt.
notifyAdoptedObjectComment :: Text
notifyAdoptedObjectComment :: Text
notifyAdoptedObjectComment = Text
notifyObjectCommentPrefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"adopted"

-- | Ownership marker stamped on every event-streaming function and trigger arbiter
-- installs. Bump the version whenever 'createEventStreamingTriggersSQL' changes.
eventStreamingObjectComment :: Text
eventStreamingObjectComment :: Text
eventStreamingObjectComment = Text
eventStreamingObjectCommentPrefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"v1"

-- | The marker prefix identifying an event-streaming object as arbiter's, across versions.
eventStreamingObjectCommentPrefix :: Text
eventStreamingObjectCommentPrefix :: Text
eventStreamingObjectCommentPrefix = Text
"arbiter:event-stream:"

-- | Marker stamped on event-streaming objects installed before arbiter marked them.
-- See 'notifyAdoptedObjectComment'.
eventStreamingAdoptedObjectComment :: Text
eventStreamingAdoptedObjectComment :: Text
eventStreamingAdoptedObjectComment = Text
eventStreamingObjectCommentPrefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"adopted"

-- | Any schema-qualified table: @qualifiedTable "arbiter" "arbiter_workers"@ -> @"arbiter"."arbiter_workers"@
qualifiedTable :: SchemaName -> TableName -> Text
qualifiedTable :: Text -> Text -> Text
qualifiedTable Text
schemaName Text
tableName = Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
tableName

-- | Qualified table name: @jobQueueTable "arbiter" "email_jobs"@ -> @"arbiter"."email_jobs"@
jobQueueTable :: SchemaName -> TableName -> Text
jobQueueTable :: Text -> Text -> Text
jobQueueTable = Text -> Text -> Text
qualifiedTable

-- | Qualified DLQ table name: @jobQueueDLQTable "arbiter" "email_jobs"@ -> @"arbiter"."email_jobs_dlq"@
jobQueueDLQTable :: SchemaName -> TableName -> Text
jobQueueDLQTable :: Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName = Text -> Text -> Text
qualifiedTable Text
schemaName (Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
dlqSuffix)

-- | Qualified archive table name: @jobQueueArchiveTable "arbiter" "email_jobs"@ -> @"arbiter"."email_jobs_archive"@
jobQueueArchiveTable :: SchemaName -> TableName -> Text
jobQueueArchiveTable :: Text -> Text -> Text
jobQueueArchiveTable Text
schemaName Text
tableName = Text -> Text -> Text
qualifiedTable Text
schemaName (Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
archiveSuffix)

-- | Backfill NULL @max_attempts@ to the default and set the column default.
-- The column stays nullable for rolling deploys.
setMaxAttemptsDefaultSQL :: SchemaName -> TableName -> Text
setMaxAttemptsDefaultSQL :: Text -> Text -> Text
setMaxAttemptsDefaultSQL Text
schemaName Text
tableName =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dma :: Text
dma = String -> Text
T.pack (Int32 -> String
forall a. Show a => a -> String
show Int32
defaultMaxAttempts)
   in [text|
        UPDATE ${tbl} SET max_attempts = ${dma} WHERE max_attempts IS NULL;
        ALTER TABLE ${tbl} ALTER COLUMN max_attempts SET DEFAULT ${dma};
      |]

-- | Qualified results table name: @jobQueueResultsTable "arbiter" "email_jobs"@ -> @"arbiter"."email_jobs_results"@
jobQueueResultsTable :: Text -> Text -> Text
jobQueueResultsTable :: Text -> Text -> Text
jobQueueResultsTable Text
schemaName Text
tableName = Text -> Text -> Text
qualifiedTable Text
schemaName (Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
resultsSuffix)

-- | Qualified groups table name: @jobQueueGroupsTable "arbiter" "email_jobs"@ -> @"arbiter"."email_jobs_groups"@
jobQueueGroupsTable :: Text -> Text -> Text
jobQueueGroupsTable :: Text -> Text -> Text
jobQueueGroupsTable Text
schemaName Text
tableName = Text -> Text -> Text
qualifiedTable Text
schemaName (Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
groupsSuffix)

dlqSuffix, archiveSuffix, resultsSuffix, groupsSuffix :: Text
dlqSuffix :: Text
dlqSuffix = Text
"_dlq"
archiveSuffix :: Text
archiveSuffix = Text
"_archive"
resultsSuffix :: Text
resultsSuffix = Text
"_results"
groupsSuffix :: Text
groupsSuffix = Text
"_groups"

-- | A queue's own table and its companions, unqualified and unquoted, for callers that
-- match on @pg_catalog@ relnames.
queueTableNames :: TableName -> [TableName]
queueTableNames :: Text -> [Text]
queueTableNames Text
tableName = Text
tableName Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: (Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<>) [Text
dlqSuffix, Text
archiveSuffix, Text
resultsSuffix, Text
groupsSuffix]

-- | Create the schema arbiter's tables live in.
createSchemaSQL :: SchemaName -> Text
createSchemaSQL :: Text -> Text
createSchemaSQL Text
schemaName =
  Text
"CREATE SCHEMA IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"

-- | The job columns the queue and DLQ tables share. Checksummed by the create-table
-- migration. A new column ships as its own ALTER script.
jobColumns :: [Text]
jobColumns :: [Text]
jobColumns =
  [ Text
"  id BIGSERIAL PRIMARY KEY,"
  , Text
"  payload JSONB NOT NULL,"
  , Text
"  group_key TEXT,"
  , Text
"  inserted_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),"
  , Text
"  updated_at TIMESTAMPTZ,"
  , Text
"  last_attempted_at TIMESTAMPTZ,"
  , Text
"  not_visible_until TIMESTAMPTZ,"
  , Text
"  attempts INT NOT NULL DEFAULT 0,"
  , Text
"  last_error TEXT,"
  , Text
"  priority INT NOT NULL DEFAULT 0,"
  , Text
"  dedup_key TEXT,"
  , Text
"  dedup_strategy TEXT,"
  , Text
"  max_attempts INT,"
  , Text
"  parent_id BIGINT,"
  , Text
"  parent_state JSONB,"
  , Text
"  suspended BOOLEAN NOT NULL DEFAULT FALSE"
  ]

-- | @ADD COLUMN IF NOT EXISTS@ over a queue's three job tables, one statement each.
addJobColumnsSQL :: Text -> Text -> [Text] -> Text
addJobColumnsSQL :: Text -> Text -> [Text] -> Text
addJobColumnsSQL Text
schemaName Text
tableName [Text]
columns =
  [Text] -> Text
T.unlines [Text -> Text
alter (Text -> Text -> Text
tbl Text
schemaName Text
tableName) | Text -> Text -> Text
tbl <- [Text -> Text -> Text
jobQueueTable, Text -> Text -> Text
jobQueueDLQTable, Text -> Text -> Text
jobQueueArchiveTable]]
  where
    alter :: Text -> Text
alter Text
table = Text
"ALTER TABLE " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
table Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> [Text] -> Text
T.intercalate Text
", " ((Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map Text -> Text
forall {a}. (Semigroup a, IsString a) => a -> a
addColumn [Text]
columns) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"
    addColumn :: a -> a
addColumn a
column = a
"ADD COLUMN IF NOT EXISTS " a -> a -> a
forall a. Semigroup a => a -> a -> a
<> a
column

-- | Add the W3C trace-context columns to a queue's three job tables.
addTraceContextColumnSQL :: Text -> Text -> Text
addTraceContextColumnSQL :: Text -> Text -> Text
addTraceContextColumnSQL Text
schemaName Text
tableName =
  Text -> Text -> [Text] -> Text
addJobColumnsSQL Text
schemaName Text
tableName [Text
"traceparent TEXT", Text
"tracestate TEXT"]

-- | Add the per-claim token column to a queue's three job tables.
addClaimSeqColumnSQL :: Text -> Text -> Text
addClaimSeqColumnSQL :: Text -> Text -> Text
addClaimSeqColumnSQL Text
schemaName Text
tableName =
  Text -> Text -> [Text] -> Text
addJobColumnsSQL Text
schemaName Text
tableName [Text
"claim_seq BIGINT NOT NULL DEFAULT 0"]

-- | Add the payload variant label to a queue's three job tables.
addKindColumnSQL :: Text -> Text -> Text
addKindColumnSQL :: Text -> Text -> Text
addKindColumnSQL Text
schemaName Text
tableName =
  Text -> Text -> [Text] -> Text
addJobColumnsSQL Text
schemaName Text
tableName [Text
"kind TEXT"]

-- | 'jobColumns' for the DLQ table, with @job_id@ in place of @id@.
jobColumnsForDLQ :: Text
jobColumnsForDLQ :: Text
jobColumnsForDLQ =
  [Text] -> Text
T.unlines
    [ Text
"  id BIGSERIAL PRIMARY KEY,"
    , Text
"  failed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),"
    , Text
"  job_id BIGINT NOT NULL,"
    ]
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> [Text] -> Text
T.unlines (Int -> [Text] -> [Text]
forall a. Int -> [a] -> [a]
drop Int
1 [Text]
jobColumns)

-- | Create a queue's main job table, holding its pending and in-progress jobs.
createJobQueueTableSQL :: Text -> Text -> Text
createJobQueueTableSQL :: Text -> Text -> Text
createJobQueueTableSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE TABLE IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ("
    , [Text] -> Text
T.unlines [Text]
jobColumns
    , Text
") WITH (fillfactor = 70);"
    ]

-- | Create a queue's DLQ table, where a job that runs out of attempts lands as a full
-- snapshot plus its failure metadata.
createJobQueueDLQTableSQL :: Text -> Text -> Text
createJobQueueDLQTableSQL :: Text -> Text -> Text
createJobQueueDLQTableSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE TABLE IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ("
    , Text
jobColumnsForDLQ
    , Text
");"
    ]

-- | Archive table columns: every Job read column (@job_id@ for @id@) plus the
-- write-only @rate_limit_cost@, the completed root job's @result@, and the
-- @completed_at@/@archive_expires_at@ metadata.
jobColumnsForArchive :: Text
jobColumnsForArchive :: Text
jobColumnsForArchive =
  [Text] -> Text
T.unlines
    [ Text
"  id BIGSERIAL PRIMARY KEY,"
    , Text
"  completed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),"
    , Text
"  archive_expires_at TIMESTAMPTZ NOT NULL,"
    , Text
"  job_id BIGINT NOT NULL,"
    , Text
"  claimed_by UUID,"
    , Text
"  archive_for INT,"
    , Text
"  rate_limit_key TEXT,"
    , Text
"  rate_limit_prefix TEXT,"
    , Text
"  rate_limit_cost DOUBLE PRECISION,"
    , Text
"  concurrency_key TEXT,"
    , Text
"  concurrency_prefix TEXT,"
    , Text
"  result JSONB,"
    ]
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> [Text] -> Text
T.unlines (Int -> [Text] -> [Text]
forall a. Int -> [a] -> [a]
drop Int
1 [Text]
jobColumns)

-- | Create the completed-job archive table. The table is logged.
createJobQueueArchiveTableSQL :: Text -> Text -> Text
createJobQueueArchiveTableSQL :: Text -> Text -> Text
createJobQueueArchiveTableSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE TABLE IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueArchiveTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ("
    , Text
jobColumnsForArchive
    , Text
");"
    ]

-- | Index on archive @completed_at@ for the most-recent-first history listing.
createArchiveCompletedAtIndexSQL :: Text -> Text -> Text
createArchiveCompletedAtIndexSQL :: Text -> Text -> Text
createArchiveCompletedAtIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_archive_completed_at")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueArchiveTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (completed_at DESC);"
    ]

-- | Index on archive @archive_expires_at@. Drives the retention purge sweep.
createArchiveExpiresAtIndexSQL :: Text -> Text -> Text
createArchiveExpiresAtIndexSQL :: Text -> Text -> Text
createArchiveExpiresAtIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_archive_expires_at")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueArchiveTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (archive_expires_at);"
    ]

-- | Index on archive @job_id@ for by-id lookups (@getArchivedJobById@).
createArchiveJobIdIndexSQL :: Text -> Text -> Text
createArchiveJobIdIndexSQL :: Text -> Text -> Text
createArchiveJobIdIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_archive_job_id")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueArchiveTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (job_id);"
    ]

-- | Index on archive @parent_id@ for per-tree history lookups.
createArchiveParentIdIndexSQL :: Text -> Text -> Text
createArchiveParentIdIndexSQL :: Text -> Text -> Text
createArchiveParentIdIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_archive_parent_id")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueArchiveTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (parent_id)"
    , Text
"WHERE parent_id IS NOT NULL;"
    ]

-- | Index on archive @group_key@ for per-group history lookups.
createArchiveGroupKeyIndexSQL :: Text -> Text -> Text
createArchiveGroupKeyIndexSQL :: Text -> Text -> Text
createArchiveGroupKeyIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_archive_group_key")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueArchiveTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (group_key);"
    ]

-- | Ranking index over ready ungrouped jobs (@not_visible_until IS NULL AND NOT
-- suspended@). The claim's ordered @LIMIT@ stops the scan at the first ready rows.
createJobQueueUngroupedReadyRankingIndexSQL :: Text -> Text -> Text
createJobQueueUngroupedReadyRankingIndexSQL :: Text -> Text -> Text
createJobQueueUngroupedReadyRankingIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_ungrouped_ready_ranking")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (priority ASC, id ASC)"
    , Text
"WHERE group_key IS NULL AND not_visible_until IS NULL AND NOT suspended;"
    ]

-- | Due-finder for ungrouped parked rows. The claim range-scans it by
-- @not_visible_until <= NOW()@ for due scheduled, backoff and expired-lease jobs.
createJobQueueUngroupedDueIndexSQL :: Text -> Text -> Text
createJobQueueUngroupedDueIndexSQL :: Text -> Text -> Text
createJobQueueUngroupedDueIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_ungrouped_due")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (not_visible_until ASC)"
    , Text
"WHERE group_key IS NULL AND not_visible_until IS NOT NULL AND NOT suspended;"
    ]

-- | Replace the full ungrouped ranking index with the ready-only ranking index
-- plus the due-finder.
migrateUngroupedReadySplitIndexesSQL :: Text -> Text -> Text
migrateUngroupedReadySplitIndexesSQL :: Text -> Text -> Text
migrateUngroupedReadySplitIndexesSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"DROP INDEX IF EXISTS "
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
schemaName
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"."
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_ungrouped_ranking")
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"
    , Text -> Text -> Text
createJobQueueUngroupedReadyRankingIndexSQL Text
schemaName Text
tableName
    , Text -> Text -> Text
createJobQueueUngroupedDueIndexSQL Text
schemaName Text
tableName
    ]

-- | Index on DLQ @group_key@, for per-group failure listings.
createDLQGroupKeyIndexSQL :: Text -> Text -> Text
createDLQGroupKeyIndexSQL :: Text -> Text -> Text
createDLQGroupKeyIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_dlq_group_key")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (group_key);"
    ]

-- | Index on DLQ @failed_at@, for the most-recent-first listing.
createDLQFailedAtIndexSQL :: Text -> Text -> Text
createDLQFailedAtIndexSQL :: Text -> Text -> Text
createDLQFailedAtIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_dlq_failed_at")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (failed_at DESC);"
    ]

-- | Index on DLQ @parent_id@, for per-parent child lookups and counts.
createDLQParentIdIndexSQL :: Text -> Text -> Text
createDLQParentIdIndexSQL :: Text -> Text -> Text
createDLQParentIdIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_dlq_parent_id")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (parent_id)"
    , Text
"WHERE parent_id IS NOT NULL;"
    ]

-- | Unique index on @dedup_key@. The dedup @ON CONFLICT@ resolves against it.
createDedupKeyIndexSQL :: Text -> Text -> Text
createDedupKeyIndexSQL :: Text -> Text -> Text
createDedupKeyIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE UNIQUE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_dedup_key")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (dedup_key)"
    , Text
"WHERE dedup_key IS NOT NULL;"
    ]

-- | Partial index on @parent_id@, for per-parent child lookups.
createParentIdIndexSQL :: Text -> Text -> Text
createParentIdIndexSQL :: Text -> Text -> Text
createParentIdIndexSQL Text
schemaName Text
tableName =
  [Text] -> Text
T.unlines
    [ Text
"CREATE INDEX IF NOT EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> 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
"_parent_id")
    , Text
"ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" (parent_id)"
    , Text
"WHERE parent_id IS NOT NULL;"
    ]

-- | Create a queue's results table, one row per child keyed by @(parent_id, child_id)@.
-- Its foreign key cascades. Acking the parent clears them.
createResultsTableSQL :: Text -> Text -> Text
createResultsTableSQL :: Text -> Text -> Text
createResultsTableSQL Text
schemaName Text
tableName =
  let resultsTbl :: Text
resultsTbl = Text -> Text -> Text
jobQueueResultsTable Text
schemaName Text
tableName
      mainTbl :: Text
mainTbl = Text -> Text -> Text
jobQueueTable 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
resultsTbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ("
        , Text
"  parent_id BIGINT NOT NULL REFERENCES " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
mainTbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"(id) ON DELETE CASCADE,"
        , Text
"  child_id BIGINT NOT NULL,"
        , Text
"  result JSONB NOT NULL,"
        , Text
"  PRIMARY KEY (parent_id, child_id)"
        , Text
");"
        ]

-- ---------------------------------------------------------------------------
-- Groups Maintenance Triggers
-- ---------------------------------------------------------------------------

-- | The qualified @<baseName>_{insert,delete,update}@ maintenance-function names.
maintenanceFunctionNames :: Text -> Text -> (Text, Text, Text)
maintenanceFunctionNames :: Text -> Text -> (Text, Text, Text)
maintenanceFunctionNames Text
schemaName Text
baseName =
  (Text -> Text
func Text
"_insert", Text -> Text
func Text
"_delete", Text -> Text
func Text
"_update")
  where
    func :: Text -> Text
func Text
suffix = Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier (Text
baseName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
suffix)

-- | One statement-level AFTER trigger. Drops then recreates, wiring the
-- @<baseName><suffix>@ function over @tbl@ with the given event and REFERENCING clause.
statementTriggerSQL :: Text -> Text -> Text -> Text -> Text -> Text -> Text
statementTriggerSQL :: Text -> Text -> Text -> Text -> Text -> Text -> Text
statementTriggerSQL Text
schemaName Text
tbl Text
baseName Text
suffix Text
event Text
referencing =
  let func :: Text
func = Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier (Text
baseName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
suffix)
      trig :: Text
trig = Text -> Text
quoteIdentifier (Text
baseName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
suffix)
   in Text -> [Text] -> Text
T.intercalate
        Text
"\n"
        [ Text
"DROP TRIGGER IF EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
trig Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"
        , Text
"CREATE TRIGGER " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
trig
        , Text
"AFTER " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
event Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tbl
        , Text
"REFERENCING " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
referencing
        , Text
"FOR EACH STATEMENT EXECUTE FUNCTION " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
func Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"();"
        ]

-- | The 3 statement-level AFTER triggers (insert/delete/update) wiring a table's
-- maintenance functions, named @<baseName>_{insert,delete,update}@.
createMaintenanceTriggersSQL :: Text -> Text -> Text -> Text
createMaintenanceTriggersSQL :: Text -> Text -> Text -> Text
createMaintenanceTriggersSQL Text
schemaName Text
tbl Text
baseName =
  Text -> [Text] -> Text
T.intercalate
    Text
"\n\n"
    [ Text -> Text -> Text -> Text -> Text -> Text -> Text
statementTriggerSQL Text
schemaName Text
tbl Text
baseName Text
"_insert" Text
"INSERT" Text
"NEW TABLE AS new_table"
    , Text -> Text -> Text -> Text -> Text -> Text -> Text
statementTriggerSQL Text
schemaName Text
tbl Text
baseName Text
"_delete" Text
"DELETE" Text
"OLD TABLE AS old_table"
    , Text -> Text -> Text -> Text -> Text -> Text -> Text
statementTriggerSQL Text
schemaName Text
tbl Text
baseName Text
"_update" Text
"UPDATE" Text
"OLD TABLE AS old_table NEW TABLE AS new_table"
    ]
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"\n"

-- | SQL for the per-table NOTIFY function, fired once per insert statement.
-- A statement that inserted nothing notifies nothing. Channel name is quoted as
-- a string literal.
createNotifyFunctionSQL :: Text -> Text -> Text
createNotifyFunctionSQL :: Text -> Text -> Text
createNotifyFunctionSQL Text
schemaName Text
tableName =
  let functionName :: Text
functionName = Text -> Text
notifyFunctionName Text
tableName
      channelName :: Text
channelName = Text -> Text
notificationChannelForTable Text
tableName
      quotedChannel :: Text
quotedChannel = HasCallStack => Text -> Text -> Text -> Text
Text -> Text -> Text -> Text
T.replace Text
"'" Text
"''" Text
channelName -- Escape single quotes for string literal
   in [Text] -> Text
T.unlines
        [ Text
"CREATE OR REPLACE FUNCTION " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
functionName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"()"
        , Text
"RETURNS TRIGGER AS $$"
        , Text
"BEGIN"
        , Text
"  IF EXISTS (SELECT 1 FROM new_table) THEN"
        , Text
"    PERFORM pg_notify('" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
quotedChannel Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"', '');"
        , Text
"  END IF;"
        , Text
"  RETURN NULL;"
        , Text
"END;"
        , Text
"$$ LANGUAGE plpgsql;"
        , Text
"COMMENT ON FUNCTION "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
schemaName
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"."
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
functionName
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"() IS '"
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
notifyObjectComment
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"';"
        ]

-- | A table's job-arrival NOTIFY trigger. Statement-level. A batch insert notifies
-- one time.
createNotifyTriggerSQL :: Text -> Text -> Text
createNotifyTriggerSQL :: Text -> Text -> Text
createNotifyTriggerSQL Text
schemaName Text
tableName =
  let functionName :: Text
functionName = Text -> Text
notifyFunctionName Text
tableName
      trigName :: Text
trigName = Text -> Text
quoteIdentifier (Text -> Text
notifyTriggerName Text
tableName)
      tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
   in [Text] -> Text
T.unlines
        [ Text
"DROP TRIGGER IF EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
trigName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"
        , Text
"CREATE TRIGGER " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
trigName
        , Text
"AFTER INSERT ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tbl
        , Text
"REFERENCING NEW TABLE AS new_table"
        , Text
"FOR EACH STATEMENT"
        , Text
"EXECUTE FUNCTION " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
functionName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"();"
        , Text
"COMMENT ON TRIGGER " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
trigName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tbl Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" IS '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
notifyObjectComment Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"';"
        ]

-- | Drop a table's job-arrival NOTIFY trigger.
dropNotifyTriggerSQL :: Text -> Text -> Text
dropNotifyTriggerSQL :: Text -> Text -> Text
dropNotifyTriggerSQL Text
schemaName Text
tableName =
  Text
"DROP TRIGGER IF EXISTS "
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier (Text -> Text
notifyTriggerName Text
tableName)
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON "
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"

-- | Drop a table's job-arrival NOTIFY function.
dropNotifyFunctionSQL :: Text -> Text -> Text
dropNotifyFunctionSQL :: Text -> Text -> Text
dropNotifyFunctionSQL Text
schemaName Text
tableName =
  let functionName :: Text
functionName = Text -> Text
notifyFunctionName Text
tableName
   in Text
"DROP FUNCTION IF EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
functionName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"();"

-- ---------------------------------------------------------------------------
-- Event Streaming Triggers (for admin UI / SSE)
-- ---------------------------------------------------------------------------

-- | Event-streaming function that receives the logical queue name and DLQ flag
-- from each trigger. Queue names ending in @_dlq@ stay unambiguous. A lease-extend
-- update emits no event.
createEventStreamingFunctionSQL :: SchemaName -> Text
createEventStreamingFunctionSQL :: Text -> Text
createEventStreamingFunctionSQL Text
schemaName =
  let funcName :: Text
funcName = Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
eventStreamingFunctionName
   in [Text] -> Text
T.unlines
        [ Text
"CREATE OR REPLACE FUNCTION " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
funcName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"() RETURNS trigger AS $$"
        , Text
"DECLARE"
        , Text
"  event_type text;"
        , Text
"  job_id bigint;"
        , Text
"  queue_name text := TG_ARGV[0];"
        , Text
"  is_dlq boolean := TG_ARGV[1]::boolean;"
        , Text
"BEGIN"
        , Text
"  IF TG_OP = 'UPDATE' THEN"
        , Text
"    IF NEW.claimed_by IS NOT NULL AND NEW.claim_seq = OLD.claim_seq"
        , Text
"       AND NEW.not_visible_until >= OLD.not_visible_until"
        , Text
"       AND " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
forall {a}. (Semigroup a, IsString a) => a -> a
leaseStripped Text
"OLD" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" = " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
forall {a}. (Semigroup a, IsString a) => a -> a
leaseStripped Text
"NEW" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" THEN"
        , Text
"      RETURN NULL;"
        , Text
"    END IF;"
        , Text
"  END IF;"
        , Text
"  CASE TG_OP"
        , Text
"    WHEN 'INSERT' THEN"
        , Text
"      event_type := CASE WHEN is_dlq THEN 'job_dlq' ELSE 'job_inserted' END;"
        , Text
"      job_id := NEW.id;"
        , Text
"    WHEN 'UPDATE' THEN"
        , Text
"      event_type := 'job_updated';"
        , Text
"      job_id := NEW.id;"
        , Text
"    WHEN 'DELETE' THEN"
        , Text
"      event_type := 'job_deleted';"
        , Text
"      job_id := OLD.id;"
        , Text
"  END CASE;"
        , Text
""
        , Text
"  PERFORM pg_notify('" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
eventStreamingChannel Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"',"
        , Text
"    json_build_object("
        , Text
"      'event', event_type,"
        , Text
"      'table', queue_name,"
        , Text
"      'job_id', job_id"
        , Text
"    )::text);"
        , Text
"  RETURN NULL;"
        , Text
"END;"
        , Text
"$$ LANGUAGE plpgsql;"
        , Text
"COMMENT ON FUNCTION " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
funcName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"() IS '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
eventStreamingObjectComment Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"';"
        ]
  where
    leaseStripped :: a -> a
leaseStripped a
row = a
"(to_jsonb(" a -> a -> a
forall a. Semigroup a => a -> a -> a
<> a
row a -> a -> a
forall a. Semigroup a => a -> a -> a
<> a
") - 'not_visible_until' - 'updated_at')"

-- | Install event-streaming triggers with explicit logical queue metadata.
createEventStreamingTriggersSQL :: SchemaName -> TableName -> Text
createEventStreamingTriggersSQL :: Text -> Text -> Text
createEventStreamingTriggersSQL Text
schemaName Text
tableName =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      funcName :: Text
funcName = Text -> Text
quoteIdentifier Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"." Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
eventStreamingFunctionName
      triggerCall :: Text -> Text
triggerCall Text
isDLQ =
        Text
"FOR EACH ROW EXECUTE FUNCTION "
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
funcName
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"("
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteLiteral Text
tableName
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
", "
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteLiteral Text
isDLQ
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
");"
      triggerComment :: Text -> Text -> Text
triggerComment Text
trigger Text
tableRef =
        Text
"COMMENT ON TRIGGER "
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
trigger
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON "
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableRef
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" IS "
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteLiteral Text
eventStreamingObjectComment
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"
   in Text -> Text -> Text
dropEventStreamingTriggersSQL Text
schemaName Text
tableName
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> [Text] -> Text
T.unlines
          [ Text
"CREATE TRIGGER " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier (Text -> Text
eventStreamingTriggerName Text
tableName)
          , Text
"AFTER INSERT OR UPDATE OR DELETE ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tbl
          , Text -> Text
triggerCall Text
"false"
          , Text -> Text -> Text
triggerComment (Text -> Text
eventStreamingTriggerName Text
tableName) Text
tbl
          , Text
""
          , Text
"CREATE TRIGGER " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier (Text -> Text
eventStreamingDLQTriggerName Text
tableName)
          , Text
"AFTER INSERT ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
dlqTbl
          , Text -> Text
triggerCall Text
"true"
          , Text -> Text -> Text
triggerComment (Text -> Text
eventStreamingDLQTriggerName Text
tableName) Text
dlqTbl
          ]
  where
    quoteLiteral :: Text -> Text
quoteLiteral = (Text
"'" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<>) (Text -> Text) -> (Text -> Text) -> Text -> Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"'") (Text -> Text) -> (Text -> Text) -> Text -> Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HasCallStack => Text -> Text -> Text -> Text
Text -> Text -> Text -> Text
T.replace Text
"'" Text
"''"

-- | Drop current and legacy event-streaming triggers for a queue and its DLQ.
-- The shared function is dropped separately after every queue is detached.
dropEventStreamingTriggersSQL :: Text -> Text -> Text
dropEventStreamingTriggersSQL :: Text -> Text -> Text
dropEventStreamingTriggersSQL Text
schemaName Text
tableName =
  let tbl :: Text
tbl = Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName
      dlqTbl :: Text
dlqTbl = Text -> Text -> Text
jobQueueDLQTable Text
schemaName Text
tableName
      dropTrigger :: Text -> Text -> Text
dropTrigger Text
name Text
tableRef = Text
"DROP TRIGGER IF EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
name Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" ON " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tableRef Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
";"
   in [Text] -> Text
T.unlines ([Text] -> Text) -> [Text] -> Text
forall a b. (a -> b) -> a -> b
$
        ((Text, Bool) -> Text) -> [(Text, Bool)] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (\(Text
name, Bool
isDLQ) -> Text -> Text -> Text
dropTrigger Text
name (Text -> Text -> Bool -> Text
forall a. a -> a -> Bool -> a
bool Text
tbl Text
dlqTbl Bool
isDLQ)) [(Text, Bool)]
legacyEventStreamingTriggers
          [Text] -> [Text] -> [Text]
forall a. Semigroup a => a -> a -> a
<> [ Text -> Text -> Text
dropTrigger (Text -> Text
eventStreamingTriggerName Text
tableName) Text
tbl
             , Text -> Text -> Text
dropTrigger (Text -> Text
eventStreamingDLQTriggerName Text
tableName) Text
dlqTbl
             ]

-- | Drop the schema-wide event-streaming function after its triggers are detached.
dropEventStreamingFunctionSQL :: SchemaName -> Text
dropEventStreamingFunctionSQL :: Text -> Text
dropEventStreamingFunctionSQL Text
schemaName =
  Text
"DROP FUNCTION IF EXISTS "
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
schemaName
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"."
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
quoteIdentifier Text
eventStreamingFunctionName
    Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"();"