pgmq-config 0.5.0.0 → 0.6.0.0
raw patch · 11 files changed
+2417/−76 lines, 11 filesdep +directorydep ~pgmq-configdep ~pgmq-coredep ~pgmq-effectfulPVP ok
version bump matches the API change (PVP)
Dependencies added: directory
Dependency ranges changed: pgmq-config, pgmq-core, pgmq-effectful, pgmq-hasql, pgmq-migration
API changes (from Hackage documentation)
+ Pgmq.Config: [premake] :: PartitionConfig -> !Maybe Int32
+ Pgmq.Config.Types: [premake] :: PartitionConfig -> !Maybe Int32
- Pgmq.Config: PartitionConfig :: !Text -> !Text -> PartitionConfig
+ Pgmq.Config: PartitionConfig :: !Text -> !Text -> !Maybe Int32 -> PartitionConfig
- Pgmq.Config.Types: PartitionConfig :: !Text -> !Text -> PartitionConfig
+ Pgmq.Config.Types: PartitionConfig :: !Text -> !Text -> !Maybe Int32 -> PartitionConfig
Files
- CHANGELOG.md +8/−0
- pgmq-config.cabal +71/−59
- src/Pgmq/Config.hs +2/−1
- src/Pgmq/Config/Effectful.hs +1/−0
- src/Pgmq/Config/Reconcile.hs +11/−7
- src/Pgmq/Config/Types.hs +5/−1
- test/EphemeralDb.hs +30/−6
- test/Main.hs +2/−0
- test/NotifyCrashSpec.hs +10/−2
- test/PartitionSpec.hs +111/−0
- test/fixtures/pgmq-1.12.0.sql +2166/−0
CHANGELOG.md view
@@ -1,5 +1,13 @@ # Revision history for pgmq-config +## 0.6.0.0 -- 2026-09-10++Breaking: `PartitionConfig` adds `premake :: Maybe Int32`. Supply `Nothing` for the existing+server default (4), or `Just n` for an explicit count of at least 1 on PGMQ 1.13+.+Both pool and effect adapters forward the choice through the shared internal `ReconcileOps`.+Premake is creation-only: existing queue settings are neither changed nor checked for drift.+`ReconcileOps` remains internal; no external backend migration is required.+ ## 0.5.0.0 -- 2026-08-06 The reconciler told three lies about what it had done, and could be taken down at startup
pgmq-config.cabal view
@@ -1,6 +1,6 @@-cabal-version: 3.4-name: pgmq-config-version: 0.5.0.0+cabal-version: 3.4+name: pgmq-config+version: 0.6.0.0 synopsis: Declarative queue configuration for PGMQ (PostgreSQL Message Queue) @@ -19,34 +19,41 @@ reported as drift rather than repaired, since converting a queue's type would mean destroying its messages. -homepage: https://github.com/shinzui/pgmq-hs-license: MIT-license-file: LICENSE-author: Nadeem Bitar-maintainer: Nadeem Bitar-category: Database-build-type: Simple+homepage: https://github.com/shinzui/pgmq-hs+license: MIT+license-file: LICENSE+author: Nadeem Bitar+maintainer: Nadeem Bitar+category: Database+build-type: Simple extra-doc-files: CHANGELOG.md+extra-source-files: test/fixtures/pgmq-1.12.0.sql flag effectful description: Enable effectful integration- default: True- manual: False+ default: True+ manual: False common warnings ghc-options:- -Wall -Wcompat -Widentities -Wincomplete-uni-patterns- -Wincomplete-record-updates -Wredundant-constraints- -fhide-source-paths -Wmissing-export-lists -Wpartial-fields+ -Wall+ -Wcompat+ -Widentities+ -Wincomplete-uni-patterns+ -Wincomplete-record-updates+ -Wredundant-constraints+ -fhide-source-paths+ -Wmissing-export-lists+ -Wpartial-fields -Wmissing-deriving-strategies library- import: warnings+ import: warnings exposed-modules: Pgmq.Config Pgmq.Config.Types - other-modules: Pgmq.Config.Reconcile+ other-modules: Pgmq.Config.Reconcile default-extensions: DeriveGeneric DuplicateRecordFields@@ -57,37 +64,41 @@ OverloadedStrings build-depends:- , base >=4.18 && <5- , containers >=0.6 && <0.8- , generic-lens ^>=2.2 || ^>=2.3- , hasql ^>=1.10- , hasql-pool ^>=1.4- , lens ^>=5.3- , pgmq-core >=0.5 && <0.6- , pgmq-hasql >=0.5 && <0.6- , text ^>=2.1+ base >=4.18 && <5,+ containers >=0.6 && <0.8,+ generic-lens ^>=2.2 || ^>=2.3,+ hasql ^>=1.10,+ hasql-pool ^>=1.4,+ lens ^>=5.3,+ pgmq-core >=0.6 && <0.7,+ pgmq-hasql >=0.6 && <0.7,+ text ^>=2.1, if flag(effectful) exposed-modules: Pgmq.Config.Effectful build-depends:- , effectful-core ^>=2.5 || ^>=2.6- , pgmq-effectful >=0.5 && <0.6-- hs-source-dirs: src- default-language: GHC2024+ effectful-core ^>=2.5 || ^>=2.6,+ pgmq-effectful >=0.6 && <0.7,+ hs-source-dirs: src+ default-language: GHC2024 test-suite pgmq-config-test- import: warnings- default-language: GHC2024- type: exitcode-stdio-1.0- hs-source-dirs: test- main-is: Main.hs- ghc-options: -threaded -rtsopts -with-rtsopts=-N+ import: warnings+ default-language: GHC2024+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Main.hs+ ghc-options:+ -threaded+ -rtsopts+ -with-rtsopts=-N+ other-modules: ConfigSpec EphemeralDb ForeignQueueSpec NotifyCrashSpec+ PartitionSpec default-extensions: DeriveGeneric@@ -99,31 +110,32 @@ OverloadedStrings build-depends:- , aeson ^>=2.2- , base >=4.18 && <5- , bytestring >=0.11 && <0.13- , ephemeral-pg >=0.2.1- , generic-lens ^>=2.2 || ^>=2.3- , hasql- , hasql-pool ^>=1.4- , lens ^>=5.3- , pg-migrate- , pgmq-config- , pgmq-core- , pgmq-hasql- , pgmq-migration- , postgresql-libpq >=0.10.1 && <0.12- , random ^>=1.2- , tasty ^>=1.5- , tasty-hunit ^>=0.10- , text- , time ^>=1.14+ aeson ^>=2.2,+ base >=4.18 && <5,+ bytestring >=0.11 && <0.13,+ directory,+ ephemeral-pg >=0.2.1,+ generic-lens ^>=2.2 || ^>=2.3,+ hasql,+ hasql-pool ^>=1.4,+ lens ^>=5.3,+ pg-migrate,+ pgmq-config >=0.6 && <0.7,+ pgmq-core >=0.6 && <0.7,+ pgmq-hasql >=0.6 && <0.7,+ pgmq-migration >=0.6 && <0.7,+ postgresql-libpq >=0.10.1 && <0.12,+ random ^>=1.2,+ tasty ^>=1.5,+ tasty-hunit ^>=0.10,+ text,+ time ^>=1.14, -- ForeignQueueSpec asserts that both reconciler backends use the lenient -- queue listing. The effect-backed half only exists when the library's -- effectful flag is on, so it is compiled in behind the same flag. if flag(effectful)- cpp-options: -DPGMQ_EFFECTFUL+ cpp-options: -DPGMQ_EFFECTFUL build-depends:- , effectful-core ^>=2.5 || ^>=2.6- , pgmq-effectful >=0.5 && <0.6+ effectful-core ^>=2.5 || ^>=2.6,+ pgmq-effectful >=0.6 && <0.7,
src/Pgmq/Config.hs view
@@ -58,6 +58,7 @@ createQueue = Sessions.createQueue, createUnloggedQueue = Sessions.createUnloggedQueue, createPartitionedQueue = Sessions.createPartitionedQueue,+ createPartitionedQueueWithPremake = Sessions.createPartitionedQueueWithPremake, enableNotifyInsert = Sessions.enableNotifyInsert, createFifoIndex = Sessions.createFifoIndex, bindTopic = Sessions.bindTopic,@@ -94,7 +95,7 @@ -- Changing a queue's type means dropping and recreating it, destroying every -- message it holds; that is an operator's decision, not a startup task. ----- * A partitioned queue's partition interval and retention interval are never+-- * A partitioned queue's partition interval, retention interval and premake are never -- compared, because @pgmq.list_queues()@ does not report them. Only the -- three-way shape is checked. --
src/Pgmq/Config/Effectful.hs view
@@ -29,6 +29,7 @@ createQueue = Eff.createQueue, createUnloggedQueue = Eff.createUnloggedQueue, createPartitionedQueue = Eff.createPartitionedQueue,+ createPartitionedQueueWithPremake = Eff.createPartitionedQueueWithPremake, enableNotifyInsert = Eff.enableNotifyInsert, createFifoIndex = Eff.createFifoIndex, bindTopic = Eff.bindTopic,
src/Pgmq/Config/Reconcile.hs view
@@ -45,6 +45,7 @@ createQueue :: QueueName -> m (), createUnloggedQueue :: QueueName -> m (), createPartitionedQueue :: StmtTypes.CreatePartitionedQueue -> m (),+ createPartitionedQueueWithPremake :: StmtTypes.CreatePartitionedQueue -> Int32 -> m (), enableNotifyInsert :: StmtTypes.EnableNotifyInsert -> m (), createFifoIndex :: QueueName -> m (), bindTopic :: StmtTypes.BindTopic -> m (),@@ -128,12 +129,15 @@ UnloggedQueue -> (ops ^. #createUnloggedQueue) qn PartitionedQueue pc ->- (ops ^. #createPartitionedQueue)- StmtTypes.CreatePartitionedQueue- { queueName = qn,- partitionInterval = pc ^. #partitionInterval,- retentionInterval = pc ^. #retentionInterval- }+ let query =+ StmtTypes.CreatePartitionedQueue+ { queueName = qn,+ partitionInterval = pc ^. #partitionInterval,+ retentionInterval = pc ^. #retentionInterval+ }+ in case pc ^. #premake of+ Nothing -> (ops ^. #createPartitionedQueue) query+ Just n -> (ops ^. #createPartitionedQueueWithPremake) query n pure [CreatedQueue qn declaredType] -- Notification. A missing row is enabled; a row whose interval already@@ -180,7 +184,7 @@ pure (queueAction ++ notifyAction ++ fifoAction ++ bindingActions) -- | The shape a declared config asks for, reduced to what @pgmq.list_queues()@--- can actually report. Partition interval and retention are deliberately+-- can actually report. Partition interval, retention and premake are deliberately -- dropped here: the listing does not expose them, so they are not drift-checked. declaredShape :: QueueType -> ObservedQueueType declaredShape StandardQueue = ObservedStandard
src/Pgmq/Config/Types.hs view
@@ -59,7 +59,11 @@ -- | Configuration for a partitioned queue. data PartitionConfig = PartitionConfig { partitionInterval :: !Text,- retentionInterval :: !Text+ retentionInterval :: !Text,+ -- | Creation only: Nothing uses the server default via the legacy call;+ -- Just requires PGMQ 1.13+ and a count of at least one. Existing queues+ -- are neither checked nor reconfigured for this setting.+ premake :: !(Maybe Int32) } deriving stock (Generic, Show)
test/EphemeralDb.hs view
@@ -10,7 +10,9 @@ ) where +import Control.Monad (filterM, when) import Data.List.NonEmpty (NonEmpty (..))+import Data.Text.IO qualified as TextIO import Database.PostgreSQL.Migrate ( defaultRunOptions, migrationPlan,@@ -23,7 +25,10 @@ ) import Hasql.Pool qualified as Pool import Hasql.Pool.Config qualified as PoolConfig+import Hasql.Session qualified as Session import Pgmq.Migration qualified as Migration+import System.Directory (doesFileExist)+import System.Environment (lookupEnv) -- | Run an action with a temporary PostgreSQL database that has pgmq schema installed withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a)@@ -35,9 +40,28 @@ PoolConfig.staticConnectionSettings connSettings ] pool <- Pool.acquire poolConfig- component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations- plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))- installResult <- runMigrationPlan defaultRunOptions connSettings plan- case installResult of- Left migrationErr -> error $ "Migration failed: " <> show migrationErr- Right _ -> action pool+ version <- lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+ case version of+ Just "1.12.0" -> do+ paths <- filterM doesFileExist ["test/fixtures/pgmq-1.12.0.sql", "pgmq-config/test/fixtures/pgmq-1.12.0.sql"]+ path <- case paths of+ candidate : _ -> pure candidate+ [] -> error "Missing packaged PGMQ 1.12.0 test fixture"+ sql <- TextIO.readFile path+ Pool.use pool (Session.script sql) >>= either (error . show) pure+ Nothing -> installNative connSettings+ Just "1.13.0" -> installNative connSettings+ Just invalid -> error ("Invalid PGMQ_TEST_SCHEMA_VERSION: " <> invalid)+ required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+ -- Required runs fail for absent or unusable pg_partman, not just missing metadata.+ partman <- Pool.use pool (Session.script "CREATE SCHEMA IF NOT EXISTS partman; CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman")+ case partman of+ Left err -> when required (error ("PGMQ_REQUIRE_PARTMAN=1: " <> show err))+ Right () -> pure ()+ action pool+ where+ installNative connSettings = do+ component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations+ plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))+ installResult <- runMigrationPlan defaultRunOptions connSettings plan+ either (error . ("Migration failed: " <>) . show) (const (pure ())) installResult
test/Main.hs view
@@ -6,6 +6,7 @@ import EphemeralDb (withPgmqDb) import ForeignQueueSpec qualified import NotifyCrashSpec qualified+import PartitionSpec qualified import Test.Tasty (defaultMain, testGroup) main :: IO ()@@ -15,6 +16,7 @@ testGroup "pgmq-config" [ ConfigSpec.tests pool,+ PartitionSpec.tests, -- NotifyCrashSpec manages its own PostgreSQL instance: it crashes -- the server, which the shared pool above could not survive. NotifyCrashSpec.tests,
test/NotifyCrashSpec.hs view
@@ -144,7 +144,8 @@ bracket (acquirePool db1) Pool.release $ \pool -> do throttlesAfterCrash <- listThrottleNames pool triggers <- runSession pool (Session.statement (queueTableName qn) insertTriggerCount)- metrics <- runSession pool (Sessions.queueMetrics qn)+ -- Keep the crash regression independent of the expanding metrics record.+ queueLength <- runSession pool (Session.statement (queueNameToText qn) queueLengthStatement) let channel = notifyChannelName qn notified <- withListener db1 channel $ \conn -> do@@ -160,7 +161,7 @@ obsChannel = channel, obsThrottlesAfterCrash = throttlesAfterCrash, obsTriggersAfterCrash = triggers,- obsQueueLengthAfterCrash = metrics ^. #queueLength,+ obsQueueLengthAfterCrash = queueLength, obsNotifyAfterCrash = LibPQ.notifyRelname <$> notified, obsThrottlesAfterReconcile = throttlesAfterReconcile }@@ -235,6 +236,13 @@ queueTableName qn = "q_" <> T.toLower (queueNameToText qn) -- | How many insert-notification triggers exist on the given @pgmq@ table.+queueLengthStatement :: Statement Text Int64+queueLengthStatement =+ preparable+ "SELECT queue_length FROM pgmq.metrics($1)"+ (E.param (E.nonNullable E.text))+ (D.singleRow (D.column (D.nonNullable D.int8)))+ insertTriggerCount :: Statement Text Int64 insertTriggerCount = preparable sql encoder decoder where
+ test/PartitionSpec.hs view
@@ -0,0 +1,111 @@+{-# LANGUAGE CPP #-}++module PartitionSpec (tests) where++import Control.Exception (finally)+import Control.Monad (forM_)+import Data.Int (Int32)+import Data.List (isInfixOf)+import EphemeralDb (withPgmqDb)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement (preparable)+import Pgmq.Config+import Pgmq.Types (QueueName, parseQueueName, queueNameToText)+import System.Environment (lookupEnv)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))+#ifdef PGMQ_EFFECTFUL+import Effectful (runEff)+import Effectful.Error.Static (runError)+import Pgmq.Config.Effectful (ensureQueuesReportEff)+import Pgmq.Effectful (PgmqRuntimeError, runPgmq)+#endif++tests :: TestTree+tests = testGroup "Partition compatibility" [backendTests backend | backend <- backends]++backends :: [Bool]+#ifdef PGMQ_EFFECTFUL+backends = [False, True]+#else+backends = [False]+#endif++reconcile :: Bool -> Pool.Pool -> [QueueConfig] -> IO (Either String [ReconcileAction])+#ifdef PGMQ_EFFECTFUL+reconcile True pool configs = either (Left . show) Right <$> runEff (runError @PgmqRuntimeError (runPgmq pool (ensureQueuesReportEff configs)))+#endif+reconcile _ pool configs = either (Left . show) Right <$> Pool.use pool (ensureQueuesReport configs)++backendTests :: Bool -> TestTree+backendTests effectful =+ testGroup+ (if effectful then "effectful" else "direct")+ $ [ testCase ("creation and skip " <> show requested) $ isolated $ \pool -> withPartman pool $ do+ stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+ qn <- right (parseQueueName "partition_config")+ let config n = partitionedQueue qn (PartitionConfig "10" "100" n)+ result <- reconcile effectful pool [config requested]+ if stock && requested /= Nothing+ then case result of+ Left err -> assertBool "explicit premake requires 1.13" ("42883" `isInfixOf` err)+ Right _ -> assertFailure "unsupported premake succeeded"+ else do+ actions <- right result+ case actions of+ [CreatedQueue q _] -> q @?= qn+ other -> assertFailure ("expected creation: " <> show other)+ let expected = maybe 4 id requested+ counts pool qn >>= (@?= [expected, expected])+ -- Turn every creation call into an error, proving the existing-queue+ -- path makes no call even when a different (invalid) count is requested.+ let signature = "queue_name text, partition_interval text DEFAULT '10000', retention_interval text DEFAULT '100000'" <> if stock then "" else ", premake integer DEFAULT 4"+ session pool (Session.script ("CREATE OR REPLACE FUNCTION pgmq.create_partitioned(" <> signature <> ") RETURNS void LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'unexpected partition creation'; END $$"))+ skipped <- right =<< reconcile effectful pool [config (Just 0)]+ case skipped of+ [SkippedQueue q] -> q @?= qn+ other -> assertFailure ("expected skip: " <> show other)+ counts pool qn >>= (@?= [expected, expected])+ | requested <- [Nothing, Just 2]+ ]+ ++ [ testCase "zero and negative counts fail on new queues" $ forM_ [0, -1] $ \n -> isolated $ \pool -> withPartman pool $ do+ stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+ qn <- right (parseQueueName "invalid_partition_config")+ result <- reconcile effectful pool [partitionedQueue qn (PartitionConfig "10" "100" (Just n))]+ case result of+ Left err -> assertBool "server error retained" ((if stock then "42883" else "premake must be at least 1") `isInfixOf` err)+ Right _ -> assertFailure "invalid premake succeeded"+ ]++counts :: Pool.Pool -> QueueName -> IO [Int32]+counts pool qn =+ session pool $+ Session.statement (queueNameToText qn) $+ preparable+ "select premake from partman.part_config where parent_table in ('pgmq.q_' || $1, 'pgmq.a_' || $1) order by parent_table"+ (E.param (E.nonNullable E.text))+ (D.rowList (D.column (D.nonNullable D.int4)))++session :: Pool.Pool -> Session.Session a -> IO a+session pool action = right =<< Pool.use pool action++right :: (Show e) => Either e a -> IO a+right = either (\err -> assertFailure (show err) >> fail "unreachable") pure++isolated :: (Pool.Pool -> IO ()) -> IO ()+isolated action = right =<< withPgmqDb (\pool -> action pool `finally` Pool.release pool)++withPartman :: Pool.Pool -> IO () -> IO ()+withPartman pool action = do+ available <-+ session pool $+ Session.statement () $+ preparable+ "select exists (select from pg_extension where extname = 'pg_partman')"+ E.noParams+ (D.singleRow (D.column (D.nonNullable D.bool)))+ required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+ if available then action else if required then assertFailure "pg_partman required" else putStrLn "SKIPPED: pg_partman is unavailable"
+ test/fixtures/pgmq-1.12.0.sql view
@@ -0,0 +1,2166 @@+------------------------------------------------------------+-- Schema, tables, records, privileges, indexes, etc+------------------------------------------------------------+-- When installed as an extension, we don't need to create the `pgmq` schema+-- because it is automatically created by postgres due to being declared in+-- the extension control file+DO+$$+BEGIN+ IF (SELECT NOT EXISTS( SELECT 1 FROM pg_extension WHERE extname = 'pgmq')) THEN+ CREATE SCHEMA IF NOT EXISTS pgmq;+ END IF;+END+$$;++-- Table where queues and metadata about them is stored+CREATE TABLE IF NOT EXISTS pgmq.meta (+ queue_name VARCHAR UNIQUE NOT NULL,+ is_partitioned BOOLEAN NOT NULL,+ is_unlogged BOOLEAN NOT NULL,+ created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL+);++-- Grant permission to pg_monitor to all tables and sequences+-- These grants are intentionally placed here (after creating `pgmq.meta` but before creating other tables). This+-- allows the `pg_dump` output for a fresh installation to match the output for an installation that followed the+-- upgrade path.+GRANT USAGE ON SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pg_monitor;++-- Table to track notification throttling for queues+CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.notify_insert_throttle (+ queue_name VARCHAR UNIQUE NOT NULL -- Queue name (without 'q_' prefix)+ CONSTRAINT notify_insert_throttle_meta_queue_name_fk+ REFERENCES pgmq.meta (queue_name)+ ON DELETE CASCADE,+ throttle_interval_ms INTEGER NOT NULL DEFAULT 0, -- Min milliseconds between notifications (0 = no throttling)+ last_notified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT to_timestamp(0) -- Timestamp of last sent notification+);++CREATE INDEX IF NOT EXISTS idx_notify_throttle_active+ ON pgmq.notify_insert_throttle (queue_name, last_notified_at)+ WHERE throttle_interval_ms > 0;++CREATE TABLE IF NOT EXISTS pgmq.topic_bindings+(+ pattern text NOT NULL, -- Wildcard pattern for routing key matching (* = one segment, # = zero or more segments)+ queue_name text NOT NULL -- Name of the queue that receives messages when pattern matches+ CONSTRAINT topic_bindings_meta_queue_name_fk+ REFERENCES pgmq.meta (queue_name)+ ON DELETE CASCADE,+ bound_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, -- Timestamp when the binding was created+ compiled_regex text GENERATED ALWAYS AS (+ -- Pre-compile the pattern to regex for faster matching+ -- This avoids runtime compilation on every send_topic call+ '^' ||+ replace(+ replace(+ regexp_replace(pattern, '([.+?{}()|\[\]\\^$])', '\\\1', 'g'),+ '*', '[^.]+'+ ),+ '#', '.*'+ ) || '$'+ ) STORED, -- Computed column: stores the compiled regex pattern+ CONSTRAINT topic_bindings_unique_pattern_queue UNIQUE (pattern, queue_name)+);++-- Create covering index for better performance when scanning patterns+-- Includes queue_name and compiled_regex to allow index-only scans (no table access needed)+CREATE INDEX IF NOT EXISTS idx_topic_bindings_covering ON pgmq.topic_bindings (pattern) INCLUDE (queue_name, compiled_regex);++-- Allow the following `pgmq` tables to be dumped by `pg_dump` when pgmq is installed as an extension+DO+$$+BEGIN+ IF EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pgmq') THEN+ PERFORM pg_catalog.pg_extension_config_dump('pgmq.meta', '');+ PERFORM pg_catalog.pg_extension_config_dump('pgmq.notify_insert_throttle', '');+ PERFORM pg_catalog.pg_extension_config_dump('pgmq.topic_bindings', '');+ END IF;+END+$$;++-- This type has the shape of a message in a queue, and is often returned by+-- pgmq functions that return messages.+-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation+-- expects a specific order of fields.+CREATE TYPE pgmq.message_record AS (+ msg_id BIGINT,+ read_ct INTEGER,+ enqueued_at TIMESTAMP WITH TIME ZONE,+ last_read_at TIMESTAMP WITH TIME ZONE,+ vt TIMESTAMP WITH TIME ZONE,+ message JSONB,+ headers JSONB+);++-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation+-- expects a specific order of fields.+CREATE TYPE pgmq.queue_record AS (+ queue_name VARCHAR,+ is_partitioned BOOLEAN,+ is_unlogged BOOLEAN,+ created_at TIMESTAMP WITH TIME ZONE+);++------------------------------------------------------------+-- Functions+------------------------------------------------------------++-- prevents race conditions during queue creation by acquiring a transaction-level advisory lock+-- uses a transaction advisory lock maintain the lock until transaction commit+-- a race condition would still exist if lock was released before commit+CREATE FUNCTION pgmq.acquire_queue_lock(queue_name TEXT)+RETURNS void AS $$+BEGIN+ PERFORM pg_advisory_xact_lock(hashtext('pgmq.queue_' || queue_name));+END;+$$ LANGUAGE plpgsql;++-- read_grouped_round_robin+-- reads messages while preserving FIFO within groups and interleaving across groups (layered round-robin)+CREATE FUNCTION pgmq.read_grouped_rr(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH fifo_groups AS (+ -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility+ SELECT+ COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key,+ MIN(msg_id) AS head_msg_id+ FROM pgmq.%1$I+ GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+ ),+ eligible_groups AS (+ -- Only groups whose head message is currently visible+ -- Acquire a transaction-level advisory lock per group to prevent concurrent selection+ SELECT+ g.fifo_key,+ g.head_msg_id,+ ROW_NUMBER() OVER (ORDER BY g.head_msg_id) AS group_priority+ FROM fifo_groups g+ JOIN pgmq.%2$I h ON h.msg_id = g.head_msg_id+ WHERE h.vt <= clock_timestamp()+ AND pg_try_advisory_xact_lock(pg_catalog.hashtextextended(g.fifo_key, 0))+ ),+ available_messages AS (+ -- All currently visible messages starting at the head for each eligible group+ SELECT+ m.msg_id,+ eg.group_priority,+ ROW_NUMBER() OVER (+ PARTITION BY eg.fifo_key+ ORDER BY m.msg_id+ ) AS msg_rank_in_group+ FROM pgmq.%3$I m+ JOIN eligible_groups eg+ ON COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = eg.fifo_key+ WHERE m.vt <= clock_timestamp()+ AND m.msg_id >= eg.head_msg_id+ ),+ ordered_messages AS (+ -- Layered round-robin: take rank 1 of all groups by group_priority, then rank 2, etc.+ -- Assign selection order before locking+ SELECT msg_id, ROW_NUMBER() OVER (ORDER BY msg_rank_in_group, group_priority) as selection_order+ FROM available_messages+ ),+ selected_messages AS (+ -- Lock the messages in the correct order, preserving selection_order+ SELECT om.msg_id, om.selection_order+ FROM ordered_messages om+ JOIN pgmq.%4$I m ON m.msg_id = om.msg_id+ WHERE om.selection_order <= $1+ ORDER BY om.selection_order+ FOR UPDATE OF m SKIP LOCKED+ ),+ updated_messages AS (+ UPDATE pgmq.%5$I m+ SET+ vt = clock_timestamp() + %6$L,+ read_ct = read_ct + 1,+ last_read_at = clock_timestamp()+ FROM selected_messages sm+ WHERE m.msg_id = sm.msg_id+ AND m.vt <= clock_timestamp() -- final guard to avoid duplicate reads under races+ RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers, sm.selection_order+ )+ SELECT msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers+ FROM updated_messages+ ORDER BY selection_order;+ $QUERY$,+ qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt)+ );+ RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_rr_with_poll+-- reads messages using round-robin layering across groups, with polling support+CREATE FUNCTION pgmq.read_grouped_rr_with_poll(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER,+ max_poll_seconds INTEGER DEFAULT 5,+ poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ r pgmq.message_record;+ stop_at TIMESTAMP;+BEGIN+ stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+ LOOP+ IF (SELECT clock_timestamp() >= stop_at) THEN+ RETURN;+ END IF;++ FOR r IN+ SELECT * FROM pgmq.read_grouped_rr(queue_name, vt, qty)+ LOOP+ RETURN NEXT r;+ END LOOP;+ IF FOUND THEN+ RETURN;+ ELSE+ PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+ END IF;+ END LOOP;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_head: read the head of N different FIFO groups in a single operation.+-- This supports horizontal scaling by processing groups in parallel while ensuring message ordering is preserved per group.+CREATE FUNCTION pgmq.read_grouped_head(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH fifo_groups AS (+ -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility+ SELECT+ COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key,+ MIN(msg_id) AS head_msg_id+ FROM pgmq.%1$I+ GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+ ),+ selected_messages AS (+ -- Take at most 1 message per group+ SELECT g.head_msg_id msg_id+ FROM fifo_groups g+ JOIN pgmq.%1$I q ON q.msg_id = g.head_msg_id+ WHERE q.vt <= clock_timestamp()+ ORDER BY q.msg_id+ LIMIT $1+ FOR UPDATE SKIP LOCKED+ )+ UPDATE pgmq.%1$I m+ SET+ vt = clock_timestamp() + %2$L,+ read_ct = read_ct + 1,+ last_read_at = clock_timestamp()+ FROM selected_messages sm+ WHERE m.msg_id = sm.msg_id+ RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+ $QUERY$,+ qtable, make_interval(secs => vt)+ );+ RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_head_with_poll+-- reads the head of N different FIFO groups in a single operation, with polling support+CREATE FUNCTION pgmq.read_grouped_head_with_poll(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER,+ max_poll_seconds INTEGER DEFAULT 5,+ poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ r pgmq.message_record;+ stop_at TIMESTAMPTZ;+BEGIN+ stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+ LOOP+ IF clock_timestamp() >= stop_at THEN+ RETURN;+ END IF;++ FOR r IN+ SELECT * FROM pgmq.read_grouped_head(queue_name, vt, qty)+ LOOP+ RETURN NEXT r;+ END LOOP;+ IF FOUND THEN+ RETURN;+ ELSE+ PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+ END IF;+ END LOOP;+END;+$$ LANGUAGE plpgsql;++-- a helper to format table names and check for invalid characters+CREATE FUNCTION pgmq.format_table_name(queue_name text, prefix text)+RETURNS TEXT AS $$+BEGIN+ IF queue_name ~ '\$|;|--|'''+ THEN+ RAISE EXCEPTION 'queue name contains invalid characters: $, ;, --, or \''';+ END IF;+ RETURN lower(prefix || '_' || queue_name);+END;+$$ LANGUAGE plpgsql;++-- read+-- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER,+ conditional JSONB DEFAULT '{}'+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH cte AS+ (+ SELECT msg_id+ FROM pgmq.%I+ WHERE vt <= clock_timestamp() AND CASE+ WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer+ ELSE 1+ END = 1+ ORDER BY msg_id ASC+ LIMIT $1+ FOR UPDATE SKIP LOCKED+ )+ UPDATE pgmq.%I m+ SET+ last_read_at = clock_timestamp(),+ vt = clock_timestamp() + %L,+ read_ct = read_ct + 1+ FROM cte+ WHERE m.msg_id = cte.msg_id+ RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+ $QUERY$,+ qtable, conditional, qtable, make_interval(secs => vt)+ );+ RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped+-- reads messages with AWS SQS FIFO-style batch retrieval behavior+-- attempts to return as many messages as possible from the same message group+CREATE FUNCTION pgmq.read_grouped(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH fifo_groups AS (+ -- Find the minimum msg_id for each FIFO group that's ready to be processed+ SELECT+ COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') as fifo_key,+ MIN(msg_id) as min_msg_id+ FROM pgmq.%I+ WHERE vt <= clock_timestamp()+ GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+ ),+ locked_groups AS (+ -- Lock the first available message in each FIFO group+ SELECT+ m.msg_id,+ fg.fifo_key+ FROM pgmq.%I m+ INNER JOIN fifo_groups fg ON+ COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = fg.fifo_key+ AND m.msg_id = fg.min_msg_id+ WHERE m.vt <= clock_timestamp()+ ORDER BY m.msg_id ASC+ FOR UPDATE SKIP LOCKED+ ),+ group_priorities AS (+ -- Assign priority to groups based on their oldest message+ SELECT+ fifo_key,+ msg_id as min_msg_id,+ ROW_NUMBER() OVER (ORDER BY msg_id) as group_priority+ FROM locked_groups+ ),+ filtered_groups as (+ SELECT * FROM group_priorities gp+ WHERE NOT EXISTS (+ -- Ensure no earlier message in this group is currently being processed+ SELECT 1+ FROM pgmq.%I m2+ WHERE COALESCE(m2.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key+ AND m2.vt > clock_timestamp()+ AND m2.msg_id < gp.min_msg_id+ )+ ),+ available_messages as (+ SELECT gp.fifo_key, t.msg_id,gp.group_priority,+ ROW_NUMBER() OVER (PARTITION BY gp.fifo_key ORDER BY t.msg_id) as msg_rank_in_group+ FROM filtered_groups gp+ CROSS JOIN LATERAL (+ SELECT *+ FROM pgmq.%I t+ WHERE COALESCE(t.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key+ AND t.vt <= clock_timestamp()+ ORDER BY msg_id+ LIMIT $1 -- tip to limit query impact, we know we need at most qty in each group+ ) t+ ORDER BY gp.group_priority+ ),+ batch_selection AS (+ -- Select messages to fill batch, prioritizing earliest group+ SELECT+ msg_id,+ ROW_NUMBER() OVER (ORDER BY group_priority, msg_rank_in_group) as overall_rank+ FROM available_messages+ ),+ selected_messages AS (+ -- Limit to requested quantity+ SELECT msg_id+ FROM batch_selection+ WHERE overall_rank <= $1+ ORDER BY msg_id+ FOR UPDATE SKIP LOCKED+ )+ UPDATE pgmq.%I m+ SET+ vt = clock_timestamp() + %L,+ read_ct = read_ct + 1,+ last_read_at = clock_timestamp()+ FROM selected_messages sm+ WHERE m.msg_id = sm.msg_id+ RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+ $QUERY$,+ qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt)+ );+ RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_with_poll+-- reads messages with AWS SQS FIFO-style batch retrieval behavior, with polling support+CREATE FUNCTION pgmq.read_grouped_with_poll(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER,+ max_poll_seconds INTEGER DEFAULT 5,+ poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ r pgmq.message_record;+ stop_at TIMESTAMP;+BEGIN+ stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+ LOOP+ IF (SELECT clock_timestamp() >= stop_at) THEN+ RETURN;+ END IF;++ FOR r IN+ SELECT * FROM pgmq.read_grouped(queue_name, vt, qty)+ LOOP+ RETURN NEXT r;+ END LOOP;+ IF FOUND THEN+ RETURN;+ ELSE+ PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+ END IF;+ END LOOP;+END;+$$ LANGUAGE plpgsql;++---- read_with_poll+---- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read_with_poll(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER,+ max_poll_seconds INTEGER DEFAULT 5,+ poll_interval_ms INTEGER DEFAULT 100,+ conditional JSONB DEFAULT '{}'+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ r pgmq.message_record;+ stop_at TIMESTAMP;+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+ LOOP+ IF (SELECT clock_timestamp() >= stop_at) THEN+ RETURN;+ END IF;++ sql := FORMAT(+ $QUERY$+ WITH cte AS+ (+ SELECT msg_id+ FROM pgmq.%I+ WHERE vt <= clock_timestamp() AND CASE+ WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer+ ELSE 1+ END = 1+ ORDER BY msg_id ASC+ LIMIT $1+ FOR UPDATE SKIP LOCKED+ )+ UPDATE pgmq.%I m+ SET+ last_read_at = clock_timestamp(),+ vt = clock_timestamp() + %L,+ read_ct = read_ct + 1+ FROM cte+ WHERE m.msg_id = cte.msg_id+ RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+ $QUERY$,+ qtable, conditional, qtable, make_interval(secs => vt)+ );++ FOR r IN+ EXECUTE sql USING qty+ LOOP+ RETURN NEXT r;+ END LOOP;+ IF FOUND THEN+ RETURN;+ ELSE+ PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+ END IF;+ END LOOP;+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes a message from the queue, and sends it to the archive, where its+---- saved permanently.+CREATE FUNCTION pgmq.archive(+ queue_name TEXT,+ msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+ sql TEXT;+ result BIGINT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH archived AS (+ DELETE FROM pgmq.%I+ WHERE msg_id = $1+ RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+ )+ INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers)+ SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+ FROM archived+ RETURNING msg_id;+ $QUERY$,+ qtable, atable+ );+ EXECUTE sql USING msg_id INTO result;+ RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes an array of message ids from the queue, and sends it to the archive,+---- where these messages will be saved permanently.+CREATE FUNCTION pgmq.archive(+ queue_name TEXT,+ msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH archived AS (+ DELETE FROM pgmq.%I+ WHERE msg_id = ANY($1)+ RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+ )+ INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers)+ SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+ FROM archived+ RETURNING msg_id;+ $QUERY$,+ qtable, atable+ );+ RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes a message id from the queue permanently+CREATE FUNCTION pgmq.delete(+ queue_name TEXT,+ msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+ sql TEXT;+ result BIGINT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ DELETE FROM pgmq.%I+ WHERE msg_id = $1+ RETURNING msg_id+ $QUERY$,+ qtable+ );+ EXECUTE sql USING msg_id INTO result;+ RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes an array of message ids from the queue permanently+CREATE FUNCTION pgmq.delete(+ queue_name TEXT,+ msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ DELETE FROM pgmq.%I+ WHERE msg_id = ANY($1)+ RETURNING msg_id+ $QUERY$,+ qtable+ );+ RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++-- send: actual implementation+CREATE FUNCTION pgmq.send(+ queue_name TEXT,+ msg JSONB,+ headers JSONB,+ delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ INSERT INTO pgmq.%I (vt, message, headers)+ VALUES ($2, $1, $3)+ RETURNING msg_id;+ $QUERY$,+ qtable+ );+ RETURN QUERY EXECUTE sql USING msg, delay, headers;+END;+$$ LANGUAGE plpgsql;++-- send: 2 args, no delay or headers+CREATE FUNCTION pgmq.send(+ queue_name TEXT,+ msg JSONB+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp());+$$ LANGUAGE sql;++-- send: 3 args with headers+CREATE FUNCTION pgmq.send(+ queue_name TEXT,+ msg JSONB,+ headers JSONB+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp());+$$ LANGUAGE sql;++-- send: 3 args with integer delay+CREATE FUNCTION pgmq.send(+ queue_name TEXT,+ msg JSONB,+ delay INTEGER+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- send: 3 args with timestamp+CREATE FUNCTION pgmq.send(+ queue_name TEXT,+ msg JSONB,+ delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send(queue_name, msg, NULL, delay);+$$ LANGUAGE sql;++-- send: 4 args with integer delay+CREATE FUNCTION pgmq.send(+ queue_name TEXT,+ msg JSONB,+ headers JSONB,+ delay INTEGER+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- _validate_batch_params: Private function to validate batch parameters+CREATE FUNCTION pgmq._validate_batch_params(+ msgs JSONB[],+ headers JSONB[]+) RETURNS void AS $$+BEGIN+ -- Validate that msgs is not NULL or empty+ IF msgs IS NULL OR array_length(msgs, 1) IS NULL THEN+ RAISE EXCEPTION 'msgs cannot be NULL or empty';+ END IF;++ -- Validate that headers array length matches msgs array length if headers is provided+ -- Note: array_length returns NULL for empty arrays, so we use COALESCE to treat empty arrays as length 0+ IF headers IS NOT NULL AND COALESCE(array_length(headers, 1), 0) != COALESCE(array_length(msgs, 1), 0) THEN+ RAISE EXCEPTION 'headers array length (%) must match msgs array length (%)',+ COALESCE(array_length(headers, 1), 0), COALESCE(array_length(msgs, 1), 0);+ END IF;+END;+$$ LANGUAGE plpgsql;++-- _send_batch: Private function that performs the actual batch insert without validation+CREATE FUNCTION pgmq._send_batch(+ queue_name TEXT,+ msgs JSONB[],+ headers JSONB[],+ delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ INSERT INTO pgmq.%I (vt, message, headers)+ SELECT $2, unnest($1), unnest(coalesce($3, ARRAY[]::jsonb[]))+ RETURNING msg_id;+ $QUERY$,+ qtable+ );+ RETURN QUERY EXECUTE sql USING msgs, delay, headers;+END;+$$ LANGUAGE plpgsql;++-- send_batch: Public function with validation+CREATE FUNCTION pgmq.send_batch(+ queue_name TEXT,+ msgs JSONB[],+ headers JSONB[],+ delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+BEGIN+ PERFORM pgmq._validate_batch_params(msgs, headers);+ RETURN QUERY SELECT * FROM pgmq._send_batch(queue_name, msgs, headers, delay);+END;+$$ LANGUAGE plpgsql;++-- send batch: 2 args+CREATE FUNCTION pgmq.send_batch(+ queue_name TEXT,+ msgs JSONB[]+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp());+$$ LANGUAGE sql;++-- send batch: 3 args with headers+CREATE FUNCTION pgmq.send_batch(+ queue_name TEXT,+ msgs JSONB[],+ headers JSONB[]+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp());+$$ LANGUAGE sql;++-- send batch: 3 args with integer delay+CREATE FUNCTION pgmq.send_batch(+ queue_name TEXT,+ msgs JSONB[],+ delay INTEGER+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- send batch: 3 args with timestamp+CREATE FUNCTION pgmq.send_batch(+ queue_name TEXT,+ msgs JSONB[],+ delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, delay);+$$ LANGUAGE sql;++-- send_batch: 4 args with integer delay+CREATE FUNCTION pgmq.send_batch(+ queue_name TEXT,+ msgs JSONB[],+ headers JSONB[],+ delay INTEGER+) RETURNS SETOF BIGINT AS $$+ SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- returned by pgmq.metrics() and pgmq.metrics_all+-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation+-- expects a specific order of fields.+CREATE TYPE pgmq.metrics_result AS (+ queue_name text,+ queue_length bigint,+ newest_msg_age_sec int,+ oldest_msg_age_sec int,+ total_messages bigint,+ scrape_time timestamp with time zone,+ queue_visible_length bigint+);++-- get metrics for a single queue+CREATE FUNCTION pgmq.metrics(queue_name TEXT)+RETURNS pgmq.metrics_result AS $$+DECLARE+ result_row pgmq.metrics_result;+ query TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ query := FORMAT(+ $QUERY$+ WITH q_summary AS (+ SELECT+ count(*) as queue_length,+ count(CASE WHEN vt <= NOW() THEN 1 END) as queue_visible_length,+ EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec,+ EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec,+ NOW() as scrape_time+ FROM pgmq.%I+ ),+ all_metrics AS (+ SELECT CASE+ WHEN is_called THEN last_value ELSE 0+ END as total_messages+ FROM pgmq.%I+ )+ SELECT+ %L as queue_name,+ q_summary.queue_length,+ q_summary.newest_msg_age_sec,+ q_summary.oldest_msg_age_sec,+ all_metrics.total_messages,+ q_summary.scrape_time,+ q_summary.queue_visible_length+ FROM q_summary, all_metrics+ $QUERY$,+ qtable, qtable || '_msg_id_seq', queue_name+ );+ EXECUTE query INTO result_row;+ RETURN result_row;+END;+$$ LANGUAGE plpgsql;++-- get metrics for all queues+CREATE FUNCTION pgmq."metrics_all"()+RETURNS SETOF pgmq.metrics_result AS $$+DECLARE+ row_name RECORD;+ result_row pgmq.metrics_result;+BEGIN+ FOR row_name IN SELECT queue_name FROM pgmq.meta LOOP+ result_row := pgmq.metrics(row_name.queue_name);+ RETURN NEXT result_row;+ END LOOP;+END;+$$ LANGUAGE plpgsql;++-- list queues+CREATE FUNCTION pgmq."list_queues"()+RETURNS SETOF pgmq.queue_record AS $$+BEGIN+ RETURN QUERY SELECT * FROM pgmq.meta;+END+$$ LANGUAGE plpgsql;++-- purge queue, deleting all entries in it.+CREATE OR REPLACE FUNCTION pgmq."purge_queue"(queue_name TEXT)+RETURNS BIGINT AS $$+DECLARE+ deleted_count INTEGER;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ -- Get the row count before truncating+ EXECUTE format('SELECT count(*) FROM pgmq.%I', qtable) INTO deleted_count;++ -- Use TRUNCATE for better performance on large tables+ EXECUTE format('TRUNCATE TABLE pgmq.%I', qtable);++ -- Return the number of purged rows+ RETURN deleted_count;+END+$$ LANGUAGE plpgsql;++-- unassign archive, so it can be kept when a queue is deleted+CREATE FUNCTION pgmq."detach_archive"(queue_name TEXT)+RETURNS VOID AS $$+DECLARE+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+ RAISE WARNING 'detach_archive(queue_name) is deprecated and is a no-op. It will be removed in PGMQ v2.0. Archive tables are no longer member objects.';+END+$$ LANGUAGE plpgsql;++-- pop: implementation+CREATE FUNCTION pgmq.pop(queue_name TEXT, qty INTEGER DEFAULT 1)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ result pgmq.message_record;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH cte AS+ (+ SELECT msg_id+ FROM pgmq.%I+ WHERE vt <= clock_timestamp()+ ORDER BY msg_id ASC+ LIMIT $1+ FOR UPDATE SKIP LOCKED+ )+ DELETE from pgmq.%I+ WHERE msg_id IN (select msg_id from cte)+ RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+ $QUERY$,+ qtable, qtable+ );+ RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- Sets timestamp vt of a message, returns it+CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt TIMESTAMP WITH TIME ZONE)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ result pgmq.message_record;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ UPDATE pgmq.%I+ SET vt = $1+ WHERE msg_id = $2+ RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+ $QUERY$, + qtable+ );+ RETURN QUERY EXECUTE sql USING vt, msg_id;+END;+$$ LANGUAGE plpgsql;++-- Sets integer vt of a message, returns it+CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt INTEGER)+RETURNS SETOF pgmq.message_record AS $$+ SELECT * FROM pgmq.set_vt(queue_name, msg_id, clock_timestamp() + make_interval(secs => vt));+$$ LANGUAGE sql;++-- Sets timestamp vt of multiple messages, returns them+CREATE FUNCTION pgmq.set_vt(+ queue_name TEXT,+ msg_ids BIGINT[],+ vt TIMESTAMP WITH TIME ZONE+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+ sql := FORMAT(+ $QUERY$+ UPDATE pgmq.%I+ SET vt = $1+ WHERE msg_id = ANY($2)+ RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+ $QUERY$,+ qtable+ );+ RETURN QUERY EXECUTE sql USING vt, msg_ids;+END;+$$ LANGUAGE plpgsql;++-- Sets integer vt of multiple messages, returns them+CREATE FUNCTION pgmq.set_vt(+ queue_name TEXT,+ msg_ids BIGINT[],+ vt INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+ SELECT * FROM pgmq.set_vt(queue_name, msg_ids, clock_timestamp() + make_interval(secs => vt));+$$ LANGUAGE sql;++CREATE FUNCTION pgmq._get_pg_partman_schema()+RETURNS TEXT AS $$+ SELECT+ extnamespace::regnamespace::text+ FROM+ pg_extension+ WHERE+ extname = 'pg_partman';+$$ LANGUAGE SQL;++CREATE FUNCTION pgmq.drop_queue(queue_name TEXT, partitioned BOOLEAN)+RETURNS BOOLEAN AS $$+DECLARE+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ fq_qtable TEXT := 'pgmq.' || qtable;+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+ fq_atable TEXT := 'pgmq.' || atable;+BEGIN+ RAISE WARNING 'drop_queue(queue_name, partitioned) is deprecated and will be removed in PGMQ v2.0. Use drop_queue(queue_name) instead';++ PERFORM pgmq.drop_queue(queue_name);++ RETURN TRUE;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.drop_queue(queue_name TEXT)+RETURNS BOOLEAN AS $$+DECLARE+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ qtable_seq TEXT := qtable || '_msg_id_seq';+ fq_qtable TEXT := 'pgmq.' || qtable;+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+ fq_atable TEXT := 'pgmq.' || atable;+ partitioned BOOLEAN;+BEGIN+ PERFORM pgmq.acquire_queue_lock(queue_name);+ EXECUTE FORMAT(+ $QUERY$+ SELECT is_partitioned FROM pgmq.meta WHERE queue_name = %L+ $QUERY$,+ queue_name+ ) INTO partitioned;++ -- check if the queue exists+ IF NOT EXISTS (+ SELECT 1+ FROM information_schema.tables+ WHERE table_name = qtable and table_schema = 'pgmq'+ ) THEN+ RAISE NOTICE 'pgmq queue `%` does not exist', queue_name;+ RETURN FALSE;+ END IF;++ EXECUTE FORMAT(+ $QUERY$+ DROP TABLE IF EXISTS pgmq.%I+ $QUERY$,+ qtable+ );++ EXECUTE FORMAT(+ $QUERY$+ DROP TABLE IF EXISTS pgmq.%I+ $QUERY$,+ atable+ );++ IF EXISTS (+ SELECT 1+ FROM information_schema.tables+ WHERE table_name = 'meta' and table_schema = 'pgmq'+ ) THEN+ EXECUTE FORMAT(+ $QUERY$+ DELETE FROM pgmq.meta WHERE queue_name = %L+ $QUERY$,+ queue_name+ );+ END IF;++ IF partitioned THEN+ EXECUTE FORMAT(+ $QUERY$+ DELETE FROM %I.part_config where parent_table in (%L, %L)+ $QUERY$,+ pgmq._get_pg_partman_schema(), fq_qtable, fq_atable+ );+ END IF;++ RETURN TRUE;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.validate_queue_name(queue_name TEXT)+RETURNS void AS $$+BEGIN+ IF length(queue_name) > 47 THEN+ -- complete table identifier must be <= 63+ -- https://www.postgresql.org/docs/17/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS+ -- e.g. template_pgmq_q_my_queue is an identifier for my_queue when partitioned+ -- template_pgmq_q_ (16) + <a max length queue name> (47) = 63 + RAISE EXCEPTION 'queue name is too long, maximum length is 47 characters';+ END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._belongs_to_pgmq(table_name TEXT)+RETURNS BOOLEAN AS $$+DECLARE+ sql TEXT;+ result BOOLEAN;+BEGIN+ SELECT EXISTS (+ SELECT 1+ FROM pg_depend+ WHERE refobjid = (SELECT oid FROM pg_extension WHERE extname = 'pgmq')+ AND objid = (+ SELECT oid+ FROM pg_class+ WHERE relname = table_name+ )+ ) INTO result;+ RETURN result;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_non_partitioned(queue_name TEXT)+RETURNS void AS $$+DECLARE+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ qtable_seq TEXT := qtable || '_msg_id_seq';+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+ PERFORM pgmq.validate_queue_name(queue_name);+ PERFORM pgmq.acquire_queue_lock(queue_name);++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.%I (+ msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ last_read_at TIMESTAMP WITH TIME ZONE,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB,+ headers JSONB+ )+ $QUERY$,+ qtable+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.%I (+ msg_id BIGINT PRIMARY KEY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ last_read_at TIMESTAMP WITH TIME ZONE,+ archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB,+ headers JSONB+ );+ $QUERY$,+ atable+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC);+ $QUERY$,+ qtable || '_vt_idx', qtable+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+ $QUERY$,+ 'archived_at_idx_' || queue_name, atable+ );++ EXECUTE FORMAT(+ $QUERY$+ INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+ VALUES (%L, false, false)+ ON CONFLICT+ DO NOTHING;+ $QUERY$,+ queue_name+ );++END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_unlogged(queue_name TEXT)+RETURNS void AS $$+DECLARE+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ qtable_seq TEXT := qtable || '_msg_id_seq';+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+ PERFORM pgmq.validate_queue_name(queue_name);+ PERFORM pgmq.acquire_queue_lock(queue_name);++ EXECUTE FORMAT(+ $QUERY$+ CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.%I (+ msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ last_read_at TIMESTAMP WITH TIME ZONE,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB,+ headers JSONB+ )+ $QUERY$,+ qtable+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.%I (+ msg_id BIGINT PRIMARY KEY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ last_read_at TIMESTAMP WITH TIME ZONE,+ archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB,+ headers JSONB+ );+ $QUERY$,+ atable+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC);+ $QUERY$,+ qtable || '_vt_idx', qtable+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+ $QUERY$,+ 'archived_at_idx_' || queue_name, atable+ );++ EXECUTE FORMAT(+ $QUERY$+ INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+ VALUES (%L, false, true)+ ON CONFLICT+ DO NOTHING;+ $QUERY$,+ queue_name+ );+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._get_partition_col(partition_interval TEXT)+RETURNS TEXT AS $$+DECLARE+ num INTEGER;+BEGIN+ BEGIN+ num := partition_interval::INTEGER;+ RETURN 'msg_id';+ EXCEPTION+ WHEN others THEN+ RETURN 'enqueued_at';+ END;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._extension_exists(extension_name TEXT)+ RETURNS BOOLEAN+ LANGUAGE SQL+AS $$+SELECT EXISTS (+ SELECT 1+ FROM pg_extension+ WHERE extname = extension_name+)+$$;++CREATE FUNCTION pgmq._ensure_pg_partman_installed()+RETURNS void AS $$+BEGIN+ IF NOT pgmq._extension_exists('pg_partman') THEN+ RAISE EXCEPTION 'pg_partman is required for partitioned queues';+ END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._get_pg_partman_major_version()+RETURNS INT+LANGUAGE SQL+AS $$+ SELECT split_part(extversion, '.', 1)::INT+ FROM pg_extension+ WHERE extname = 'pg_partman'+$$;++CREATE FUNCTION pgmq.create_partitioned(+ queue_name TEXT,+ partition_interval TEXT DEFAULT '10000',+ retention_interval TEXT DEFAULT '100000'+)+RETURNS void AS $$+DECLARE+ partition_col TEXT;+ a_partition_col TEXT;+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ qtable_seq TEXT := qtable || '_msg_id_seq';+ atable TEXT := pgmq.format_table_name(queue_name, 'a');+ fq_qtable TEXT := 'pgmq.' || qtable;+ fq_atable TEXT := 'pgmq.' || atable;+BEGIN+ PERFORM pgmq.validate_queue_name(queue_name);+ PERFORM pgmq.acquire_queue_lock(queue_name);+ PERFORM pgmq._ensure_pg_partman_installed();+ SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.%I (+ msg_id BIGINT GENERATED ALWAYS AS IDENTITY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ last_read_at TIMESTAMP WITH TIME ZONE,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB,+ headers JSONB+ ) PARTITION BY RANGE (%I)+ $QUERY$,+ qtable, partition_col+ );++ -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+ -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+ EXECUTE FORMAT(+ $QUERY$+ SELECT %I.create_parent(+ p_parent_table := %L,+ p_control := %L,+ p_interval := %L,+ p_type := case+ when pgmq._get_pg_partman_major_version() = 5 then 'range'+ else 'native'+ end+ )+ $QUERY$,+ pgmq._get_pg_partman_schema(),+ fq_qtable,+ partition_col,+ partition_interval+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I);+ $QUERY$,+ qtable || '_part_idx', qtable, partition_col+ );++ EXECUTE FORMAT(+ $QUERY$+ UPDATE %I.part_config+ SET+ retention = %L,+ retention_keep_table = false,+ retention_keep_index = true,+ automatic_maintenance = 'on'+ WHERE parent_table = %L;+ $QUERY$,+ pgmq._get_pg_partman_schema(),+ retention_interval,+ 'pgmq.' || qtable+ );++ EXECUTE FORMAT(+ $QUERY$+ INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+ VALUES (%L, true, false)+ ON CONFLICT+ DO NOTHING;+ $QUERY$,+ queue_name+ );++ IF partition_col = 'enqueued_at' THEN+ a_partition_col := 'archived_at';+ ELSE+ a_partition_col := partition_col;+ END IF;++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.%I (+ msg_id BIGINT NOT NULL,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ last_read_at TIMESTAMP WITH TIME ZONE,+ archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB,+ headers JSONB+ ) PARTITION BY RANGE (%I);+ $QUERY$,+ atable, a_partition_col+ );++ -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+ -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+ EXECUTE FORMAT(+ $QUERY$+ SELECT %I.create_parent(+ p_parent_table := %L,+ p_control := %L,+ p_interval := %L,+ p_type := case+ when pgmq._get_pg_partman_major_version() = 5 then 'range'+ else 'native'+ end+ )+ $QUERY$,+ pgmq._get_pg_partman_schema(),+ fq_atable,+ a_partition_col,+ partition_interval+ );++ EXECUTE FORMAT(+ $QUERY$+ UPDATE %I.part_config+ SET+ retention = %L,+ retention_keep_table = false,+ retention_keep_index = true,+ automatic_maintenance = 'on'+ WHERE parent_table = %L;+ $QUERY$,+ pgmq._get_pg_partman_schema(),+ retention_interval,+ 'pgmq.' || atable+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+ $QUERY$,+ 'archived_at_idx_' || queue_name, atable+ );++END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create(queue_name TEXT)+RETURNS void AS $$+BEGIN+ PERFORM pgmq.create_non_partitioned(queue_name);+END;+$$ LANGUAGE plpgsql;++-- _create_fifo_index_if_not_exists+-- internal function to create GIN index on headers for better FIFO performance+CREATE OR REPLACE FUNCTION pgmq._create_fifo_index_if_not_exists(queue_name TEXT)+RETURNS void AS $$+DECLARE+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ index_name TEXT := qtable || '_fifo_idx';+BEGIN+ -- Create GIN index on headers for efficient FIFO key lookups+ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS %I ON pgmq.%I USING GIN (headers);+ $QUERY$,+ index_name, qtable+ );+END;+$$ LANGUAGE plpgsql;++-- create_fifo_index+-- creates a GIN index on the headers column to improve FIFO read performance+CREATE FUNCTION pgmq.create_fifo_index(queue_name TEXT)+RETURNS void AS $$+BEGIN+ PERFORM pgmq._create_fifo_index_if_not_exists(queue_name);+END;+$$ LANGUAGE plpgsql;++-- create_fifo_indexes_all+-- creates FIFO indexes on all existing queues+CREATE FUNCTION pgmq.create_fifo_indexes_all()+RETURNS void AS $$+DECLARE+ queue_record RECORD;+BEGIN+ FOR queue_record IN SELECT queue_name FROM pgmq.meta LOOP+ PERFORM pgmq.create_fifo_index(queue_record.queue_name);+ END LOOP;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.convert_archive_partitioned(+ table_name TEXT,+ partition_interval TEXT DEFAULT '10000',+ retention_interval TEXT DEFAULT '100000',+ leading_partition INT DEFAULT 10+)+RETURNS void AS $$+DECLARE+ a_table_name TEXT := pgmq.format_table_name(table_name, 'a');+ a_table_name_old TEXT := pgmq.format_table_name(table_name, 'a') || '_old';+ qualified_a_table_name TEXT := format('pgmq.%I', a_table_name);+ partition_col TEXT;+ a_partition_col TEXT;+BEGIN++ PERFORM c.relkind+ FROM pg_class c+ JOIN pg_namespace n ON n.oid = c.relnamespace+ WHERE c.relname = a_table_name+ AND c.relkind = 'p';++ IF FOUND THEN+ RAISE NOTICE 'Table %s is already partitioned', a_table_name;+ RETURN;+ END IF;++ PERFORM c.relkind+ FROM pg_class c+ JOIN pg_namespace n ON n.oid = c.relnamespace+ WHERE c.relname = a_table_name+ AND c.relkind = 'r';++ IF NOT FOUND THEN+ RAISE NOTICE 'Table %s does not exists', a_table_name;+ RETURN;+ END IF;++ SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;++ -- For archive tables, use archived_at for time-based partitioning+ IF partition_col = 'enqueued_at' THEN+ a_partition_col := 'archived_at';+ ELSE+ a_partition_col := partition_col;+ END IF;++ EXECUTE 'ALTER TABLE ' || qualified_a_table_name || ' RENAME TO ' || a_table_name_old;++ -- When partitioning by time (archived_at), we need to exclude constraints and indexes+ -- because the existing PRIMARY KEY on msg_id alone is incompatible with partitioning by archived_at.+ -- When partitioning by msg_id, we can keep all constraints including PRIMARY KEY.+ IF a_partition_col = 'archived_at' THEN+ EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including defaults including generated including storage including comments) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col );+ ELSE+ EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including all) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col );+ END IF;++ EXECUTE 'ALTER INDEX pgmq.archived_at_idx_' || table_name || ' RENAME TO archived_at_idx_' || table_name || '_old';+ EXECUTE 'CREATE INDEX archived_at_idx_'|| table_name || ' ON ' || qualified_a_table_name ||'(archived_at)';++ -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+ -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+ EXECUTE FORMAT(+ $QUERY$+ SELECT %I.create_parent(+ p_parent_table := %L,+ p_control := %L,+ p_interval := %L,+ p_type := case+ when pgmq._get_pg_partman_major_version() = 5 then 'range'+ else 'native'+ end+ )+ $QUERY$,+ pgmq._get_pg_partman_schema(),+ qualified_a_table_name,+ a_partition_col,+ partition_interval+ );++ EXECUTE FORMAT(+ $QUERY$+ UPDATE %I.part_config+ SET+ retention = %L,+ retention_keep_table = false,+ retention_keep_index = false,+ infinite_time_partitions = true+ WHERE+ parent_table = %L;+ $QUERY$,+ pgmq._get_pg_partman_schema(),+ retention_interval,+ qualified_a_table_name+ );+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.notify_queue_listeners()+RETURNS TRIGGER AS $$+DECLARE+ queue_name_extracted TEXT; -- Queue name extracted from trigger table name+ updated_count INTEGER; -- Number of rows updated (0 or 1)+BEGIN+ queue_name_extracted := substring(TG_TABLE_NAME from 3);++ UPDATE pgmq.notify_insert_throttle+ SET last_notified_at = clock_timestamp()+ WHERE queue_name = queue_name_extracted+ AND (+ throttle_interval_ms = 0 -- No throttling configured+ OR clock_timestamp() - last_notified_at >=+ (throttle_interval_ms * INTERVAL '1 millisecond') -- Throttle interval has elapsed+ );++ -- Check how many rows were updated (will be 0 or 1)+ GET DIAGNOSTICS updated_count = ROW_COUNT;++ IF updated_count > 0 THEN+ PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL);+ END IF;++RETURN NEW;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.enable_notify_insert(queue_name TEXT, throttle_interval_ms INTEGER DEFAULT 250)+RETURNS void AS $$+DECLARE+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ v_queue_name TEXT := queue_name;+ v_throttle_interval_ms INTEGER := throttle_interval_ms;+BEGIN+ -- Validate that throttle_interval_ms is non-negative+ IF v_throttle_interval_ms < 0 THEN+ RAISE EXCEPTION 'throttle_interval_ms must be non-negative';+ END IF;++ -- Validate that the queue table exists+ IF NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgmq' AND table_name = qtable) THEN+ RAISE EXCEPTION 'Queue "%" does not exist. Create it first using pgmq.create()', v_queue_name;+ END IF;++ PERFORM pgmq.disable_notify_insert(v_queue_name);++ INSERT INTO pgmq.notify_insert_throttle (queue_name, throttle_interval_ms)+ VALUES (v_queue_name, v_throttle_interval_ms)+ ON CONFLICT ON CONSTRAINT notify_insert_throttle_queue_name_key DO UPDATE+ SET throttle_interval_ms = EXCLUDED.throttle_interval_ms,+ last_notified_at = to_timestamp(0);++ EXECUTE FORMAT(+ $QUERY$+ CREATE CONSTRAINT TRIGGER trigger_notify_queue_insert_listeners+ AFTER INSERT ON pgmq.%I+ DEFERRABLE FOR EACH ROW+ EXECUTE PROCEDURE pgmq.notify_queue_listeners()+ $QUERY$,+ qtable+ );+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.disable_notify_insert(queue_name TEXT)+RETURNS void AS $$+DECLARE+ qtable TEXT := pgmq.format_table_name(queue_name, 'q');+ v_queue_name TEXT := queue_name;+BEGIN+ EXECUTE FORMAT(+ $QUERY$+ DROP TRIGGER IF EXISTS trigger_notify_queue_insert_listeners ON pgmq.%I;+ $QUERY$,+ qtable+ );++ DELETE FROM pgmq.notify_insert_throttle nit WHERE nit.queue_name = v_queue_name;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.list_notify_insert_throttles()+ RETURNS TABLE+ (+ queue_name text,+ throttle_interval_ms integer,+ last_notified_at TIMESTAMP WITH TIME ZONE+ )+ LANGUAGE sql+ STABLE+AS+$$+ SELECT queue_name, throttle_interval_ms, last_notified_at+ FROM pgmq.notify_insert_throttle+ ORDER BY queue_name;+$$;++CREATE OR REPLACE FUNCTION pgmq.update_notify_insert(queue_name text, throttle_interval_ms integer)+ RETURNS void+ LANGUAGE plpgsql+AS+$$+BEGIN+ IF throttle_interval_ms < 0 THEN+ RAISE EXCEPTION 'throttle_interval_ms must be non-negative, got: %', throttle_interval_ms;+ END IF;++ IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = update_notify_insert.queue_name) THEN+ RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name;+ END IF;++ IF NOT EXISTS (SELECT 1 FROM pgmq.notify_insert_throttle WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name) THEN+ RAISE EXCEPTION 'Queue "%" does not have notify_insert enabled. Enable it first using pgmq.enable_notify_insert()', queue_name;+ END IF;++ UPDATE pgmq.notify_insert_throttle+ SET throttle_interval_ms = update_notify_insert.throttle_interval_ms,+ last_notified_at = to_timestamp(0)+ WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.validate_routing_key(routing_key text)+ RETURNS boolean+ LANGUAGE plpgsql+ IMMUTABLE+AS+$$+BEGIN+ -- Valid routing key examples:+ -- "logs.error"+ -- "app.user-service.auth"+ -- "system_events.db.connection_failed"+ --+ -- Invalid routing key examples:+ -- "" - empty+ -- ".logs.error" - starts with dot+ -- "logs.error." - ends with dot+ -- "logs..error" - consecutive dots+ -- "logs.error!" - invalid character+ -- "logs error" - space not allowed+ -- "logs.*" - wildcards not allowed in routing keys++ IF routing_key IS NULL OR routing_key = '' THEN+ RAISE EXCEPTION 'routing_key cannot be NULL or empty';+ END IF;++ IF length(routing_key) > 255 THEN+ RAISE EXCEPTION 'routing_key length cannot exceed 255 characters, got % characters', length(routing_key);+ END IF;++ IF routing_key !~ '^[a-zA-Z0-9._-]+$' THEN+ RAISE EXCEPTION 'routing_key contains invalid characters. Only alphanumeric, dots, hyphens, and underscores are allowed. Got: %', routing_key;+ END IF;++ IF routing_key ~ '^\.' THEN+ RAISE EXCEPTION 'routing_key cannot start with a dot. Got: %', routing_key;+ END IF;++ IF routing_key ~ '\.$' THEN+ RAISE EXCEPTION 'routing_key cannot end with a dot. Got: %', routing_key;+ END IF;++ IF routing_key ~ '\.\.' THEN+ RAISE EXCEPTION 'routing_key cannot contain consecutive dots. Got: %', routing_key;+ END IF;++ RETURN true;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.validate_topic_pattern(pattern text)+ RETURNS boolean+ LANGUAGE plpgsql+ IMMUTABLE+AS+$$+BEGIN+ -- Valid pattern examples:+ -- "logs.*" - matches one segment after logs. (e.g., logs.error, logs.info)+ -- "logs.#" - matches one or more segments after logs. (e.g., logs.error, logs.api.error)+ -- "*.error" - matches one segment before .error (e.g., app.error, db.error)+ -- "#.error" - matches one or more segments before .error (e.g., app.error, x.y.error)+ -- "app.*.#" - mixed wildcards (one segment then one or more)+ -- "#" - catch-all pattern, matches any routing key+ --+ -- Invalid pattern examples:+ -- ".logs.*" - starts with dot+ -- "logs.*." - ends with dot+ -- "logs..error" - consecutive dots+ -- "logs.**" - consecutive stars+ -- "logs.##" - consecutive hashes+ -- "logs.*#" - adjacent wildcards+ -- "logs.error!" - invalid character++ IF pattern IS NULL OR pattern = '' THEN+ RAISE EXCEPTION 'pattern cannot be NULL or empty';+ END IF;++ IF length(pattern) > 255 THEN+ RAISE EXCEPTION 'pattern length cannot exceed 255 characters, got % characters', length(pattern);+ END IF;++ IF pattern !~ '^[a-zA-Z0-9._\-*#]+$' THEN+ RAISE EXCEPTION 'pattern contains invalid characters. Only alphanumeric, dots, hyphens, underscores, *, and # are allowed. Got: %', pattern;+ END IF;++ IF pattern ~ '^\.' THEN+ RAISE EXCEPTION 'pattern cannot start with a dot. Got: %', pattern;+ END IF;++ IF pattern ~ '\.$' THEN+ RAISE EXCEPTION 'pattern cannot end with a dot. Got: %', pattern;+ END IF;++ IF pattern ~ '\.\.' THEN+ RAISE EXCEPTION 'pattern cannot contain consecutive dots. Got: %', pattern;+ END IF;++ IF pattern ~ '\*\*' THEN+ RAISE EXCEPTION 'pattern cannot contain consecutive stars (**). Use # for multi-segment matching. Got: %', pattern;+ END IF;++ IF pattern ~ '##' THEN+ RAISE EXCEPTION 'pattern cannot contain consecutive hashes (##). A single # already matches zero or more segments. Got: %', pattern;+ END IF;++ IF pattern ~ '\*#' OR pattern ~ '#\*' THEN+ RAISE EXCEPTION 'pattern cannot contain adjacent wildcards (*# or #*). Separate wildcards with dots. Got: %', pattern;+ END IF;++ RETURN true;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.bind_topic(pattern text, queue_name text)+ RETURNS void+ LANGUAGE plpgsql+AS+$$+BEGIN+ PERFORM pgmq.validate_topic_pattern(pattern);+ IF queue_name IS NULL OR queue_name = '' THEN+ RAISE EXCEPTION 'queue_name cannot be NULL or empty';+ END IF;++ IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = bind_topic.queue_name) THEN+ RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name;+ END IF;++ INSERT INTO pgmq.topic_bindings (pattern, queue_name)+ VALUES (pattern, queue_name)+ ON CONFLICT ON CONSTRAINT topic_bindings_unique_pattern_queue DO NOTHING;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.unbind_topic(pattern text, queue_name text)+ RETURNS boolean+ LANGUAGE plpgsql+AS+$$+DECLARE+ rows_deleted integer;+BEGIN+ IF pattern IS NULL OR pattern = '' THEN+ RAISE EXCEPTION 'pattern cannot be NULL or empty';+ END IF;++ IF queue_name IS NULL OR queue_name = '' THEN+ RAISE EXCEPTION 'queue_name cannot be NULL or empty';+ END IF;++ DELETE+ FROM pgmq.topic_bindings+ WHERE topic_bindings.pattern = unbind_topic.pattern+ AND topic_bindings.queue_name = unbind_topic.queue_name;++ GET DIAGNOSTICS rows_deleted = ROW_COUNT;++ IF rows_deleted > 0 THEN+ RETURN true;+ ELSE+ RETURN false;+ END IF;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.test_routing(routing_key text)+ RETURNS TABLE+ (+ pattern text,+ queue_name text,+ compiled_regex text+ )+ LANGUAGE plpgsql+ STABLE+AS+$$+BEGIN+ PERFORM pgmq.validate_routing_key(routing_key);+ RETURN QUERY+ SELECT b.pattern,+ b.queue_name,+ b.compiled_regex+ FROM pgmq.topic_bindings b+ WHERE routing_key ~ b.compiled_regex+ ORDER BY b.pattern;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, headers jsonb, delay integer)+ RETURNS integer+ LANGUAGE plpgsql+ VOLATILE+AS+$$+DECLARE+ b RECORD;+ matched_count integer := 0;+BEGIN+ PERFORM pgmq.validate_routing_key(routing_key);++ IF msg IS NULL THEN+ RAISE EXCEPTION 'msg cannot be NULL';+ END IF;++ IF delay < 0 THEN+ RAISE EXCEPTION 'delay cannot be negative, got: %', delay;+ END IF;++ -- Filter matching patterns in SQL for better performance (uses index)+ -- Any failure will rollback the entire transaction+ FOR b IN+ SELECT DISTINCT tb.queue_name+ FROM pgmq.topic_bindings tb+ WHERE routing_key ~ tb.compiled_regex+ ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name+ LOOP+ PERFORM pgmq.send(b.queue_name, msg, headers, delay);+ matched_count := matched_count + 1;+ END LOOP;++ RETURN matched_count;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb)+ RETURNS integer+ LANGUAGE plpgsql+ VOLATILE+AS+$$+BEGIN+ RETURN pgmq.send_topic(routing_key, msg, NULL, 0);+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, delay integer)+ RETURNS integer+ LANGUAGE plpgsql+ VOLATILE+AS+$$+BEGIN+ RETURN pgmq.send_topic(routing_key, msg, NULL, delay);+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings()+ RETURNS TABLE+ (+ pattern text,+ queue_name text,+ bound_at TIMESTAMP WITH TIME ZONE,+ compiled_regex text+ )+ LANGUAGE sql+ STABLE+AS+$$+ SELECT pattern, queue_name, bound_at, compiled_regex+ FROM pgmq.topic_bindings+ ORDER BY bound_at DESC, pattern, queue_name;+$$;++CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings(queue_name text)+ RETURNS TABLE+ (+ pattern text,+ queue_name text,+ bound_at TIMESTAMP WITH TIME ZONE,+ compiled_regex text+ )+ LANGUAGE sql+ STABLE+AS+$$+ SELECT pattern, tb.queue_name, bound_at, compiled_regex+ FROM pgmq.topic_bindings tb+ WHERE tb.queue_name = list_topic_bindings.queue_name+ ORDER BY bound_at DESC, pattern;+$$;++-- send_batch_topic: Base implementation with TIMESTAMP WITH TIME ZONE delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+ routing_key text,+ msgs jsonb[],+ headers jsonb[],+ delay TIMESTAMP WITH TIME ZONE+)+ RETURNS TABLE(queue_name text, msg_id bigint)+ LANGUAGE plpgsql+ VOLATILE+AS+$$+DECLARE+ b RECORD;+BEGIN+ PERFORM pgmq.validate_routing_key(routing_key);++ -- Validate batch parameters once (not per queue)+ PERFORM pgmq._validate_batch_params(msgs, headers);++ -- Filter matching patterns in SQL for better performance (uses index)+ -- Any failure will rollback the entire transaction+ FOR b IN+ SELECT DISTINCT tb.queue_name+ FROM pgmq.topic_bindings tb+ WHERE routing_key ~ tb.compiled_regex+ ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name+ LOOP+ -- Use private _send_batch to avoid redundant validation+ RETURN QUERY+ SELECT b.queue_name, batch_result.msg_id+ FROM pgmq._send_batch(b.queue_name, msgs, headers, delay) AS batch_result(msg_id);+ END LOOP;++ RETURN;+END;+$$;++-- send_batch_topic: 2 args (routing_key, msgs)+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+ routing_key text,+ msgs jsonb[]+)+ RETURNS TABLE(queue_name text, msg_id bigint)+ LANGUAGE sql+ VOLATILE+AS+$$+ SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp());+$$;++-- send_batch_topic: 3 args with headers+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+ routing_key text,+ msgs jsonb[],+ headers jsonb[]+)+ RETURNS TABLE(queue_name text, msg_id bigint)+ LANGUAGE sql+ VOLATILE+AS+$$+ SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp());+$$;++-- send_batch_topic: 3 args with integer delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+ routing_key text,+ msgs jsonb[],+ delay integer+)+ RETURNS TABLE(queue_name text, msg_id bigint)+ LANGUAGE sql+ VOLATILE+AS+$$+ SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp() + make_interval(secs => delay));+$$;++-- send_batch_topic: 3 args with timestamp delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+ routing_key text,+ msgs jsonb[],+ delay TIMESTAMP WITH TIME ZONE+)+ RETURNS TABLE(queue_name text, msg_id bigint)+ LANGUAGE sql+ VOLATILE+AS+$$+ SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, delay);+$$;++-- send_batch_topic: 4 args with integer delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+ routing_key text,+ msgs jsonb[],+ headers jsonb[],+ delay integer+)+ RETURNS TABLE(queue_name text, msg_id bigint)+ LANGUAGE sql+ VOLATILE+AS+$$+ SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp() + make_interval(secs => delay));+$$;