pgmq-hasql 0.5.0.0 → 0.6.0.0

raw patch · 18 files changed

+2751/−128 lines, 18 filesdep +directorydep ~pgmq-coredep ~pgmq-hasqldep ~pgmq-migrationPVP ok

version bump matches the API change (PVP)

Dependencies added: directory

Dependency ranges changed: pgmq-core, pgmq-hasql, pgmq-migration

API changes (from Hackage documentation)

Files

• CHANGELOG.md +10/−0
• pgmq-hasql.cabal +66/−53
• src/Pgmq.hs +24/−0
• src/Pgmq/Hasql/Decoders.hs +1/−0
• src/Pgmq/Hasql/Encoders.hs +8/−2
• src/Pgmq/Hasql/Sessions.hs +23/−0
• src/Pgmq/Hasql/Statements/Message.hs +20/−0
• src/Pgmq/Hasql/Statements/QueueManagement.hs +10/−1
• src/Pgmq/Hasql/Statements/QueueObservability.hs +12/−2
• src/Pgmq/Hasql/Statements/Types.hs +6/−2
• test/AdvancedOpsSpec.hs +109/−2
• test/EphemeralDb.hs +33/−8
• test/Generators.hs +3/−1
• test/Main.hs +28/−22
• test/MetricsSpec.hs +95/−14
• test/QueueSpec.hs +110/−21
• test/UmbrellaExportsSpec.hs +27/−0
• test/fixtures/pgmq-1.12.0.sql +2166/−0

@@ -1,5 +1,15 @@ # Revision history for pgmq-hasql +## 0.6.0.0 -- 2026-09-10++Add grouped-head reads and polling (PGMQ 1.12+), and expose all six grouped reads and their+argument records from `Pgmq`. Add `createPartitionedQueueWithPremake` (PGMQ 1.13+), retaining+the existing three-field `CreatePartitionedQueue` and three-argument operation.++Breaking: `QueueMetrics` adds `defaultPartitionLength :: Maybe Int64`. Record constructors+must supply the field. Both metrics APIs work on 1.12 and 1.13: `Nothing` means unavailable+or inapplicable, while `Just n` estimates queue-plus-archive default-partition rows.+ ## 0.5.0.0 -- 2026-08-06  ### Breaking Changes

@@ -1,7 +1,7 @@-cabal-version:   3.4-name:            pgmq-hasql-version:         0.5.0.0-synopsis:        Hasql-based client for PGMQ (PostgreSQL Message Queue)+cabal-version: 3.4+name: pgmq-hasql+version: 0.6.0.0+synopsis: Hasql-based client for PGMQ (PostgreSQL Message Queue) description:   A Haskell client library for PGMQ (PostgreSQL Message Queue) built   on hasql. Provides queue management, message sending\/receiving,@@ -9,24 +9,31 @@   pgmq 1.5.0 through 1.11.0 features including message headers,   conditional reads, notifications, round-robin FIFO, and topic routing. -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  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     Pgmq.Hasql.Decoders@@ -51,28 +58,32 @@     OverloadedStrings    build-depends:-    , aeson              ^>=2.2-    , base               >=4.18  && <5-    , generic-lens       ^>=2.2  || ^>=2.3-    , hasql              ^>=1.10-    , hasql-transaction  ^>=1.2-    , lens               ^>=5.3-    , pgmq-core          >=0.5   && <0.6-    , template-haskell   >=2.20  && <3-    , text               ^>=2.1-    , time               ^>=1.14-    , vector             ^>=0.13+    aeson ^>=2.2,+    base >=4.18 && <5,+    generic-lens ^>=2.2 || ^>=2.3,+    hasql ^>=1.10,+    hasql-transaction ^>=1.2,+    lens ^>=5.3,+    pgmq-core >=0.6 && <0.7,+    template-haskell >=2.20 && <3,+    text ^>=2.1,+    time ^>=1.14,+    vector ^>=0.13, -  hs-source-dirs:     src-  default-language:   GHC2024+  hs-source-dirs: src+  default-language: GHC2024  test-suite pgmq-hasql-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:     AdvancedOpsSpec     AliasingSpec@@ -92,28 +103,30 @@     SchemaSpec     TestUtils     TopicSpec+    UmbrellaExportsSpec    default-extensions:     ImportQualifiedPost     OverloadedStrings    build-depends:-    , aeson-    , base              >=4.18   && <5-    , ephemeral-pg      >=0.2.1-    , hasql-    , hasql-pool        ^>=1.4-    , hedgehog          ^>=1.5-    , pg-migrate-    , pgmq-core-    , pgmq-hasql-    , pgmq-migration-    , postgresql-libpq  >=0.10.1 && <0.12-    , random            ^>=1.2-    , scientific        ^>=0.3-    , tasty             ^>=1.5-    , tasty-hedgehog    ^>=1.4-    , tasty-hunit       ^>=0.10-    , text-    , time-    , vector+    aeson,+    base >=4.18 && <5,+    directory,+    ephemeral-pg >=0.2.1,+    hasql,+    hasql-pool ^>=1.4,+    hedgehog ^>=1.5,+    pg-migrate,+    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,+    scientific ^>=0.3,+    tasty ^>=1.5,+    tasty-hedgehog ^>=1.4,+    tasty-hunit ^>=0.10,+    text,+    time,+    vector,

@@ -3,6 +3,7 @@     createQueue,     dropQueue,     createPartitionedQueue,+    createPartitionedQueueWithPremake,     createUnloggedQueue,     detachArchive, -- DEPRECATED: no-op, will be removed in pgmq 2.0 @@ -41,6 +42,20 @@     queueMetrics,     allQueueMetrics, +    -- * FIFO / Grouped Reads++    -- | Head reads (PGMQ 1.12+) lease at most one absolute head per group.+    -- An invisible head blocks its group; expiry makes that same head eligible.+    -- Polling occupies a database connection. Leases do not guarantee exactly-once processing.+    readGrouped,+    readGroupedWithPoll,+    readGroupedRoundRobin,+    readGroupedRoundRobinWithPoll,+    readGroupedHead,+    readGroupedHeadWithPoll,+    ReadGrouped (..),+    ReadGroupedWithPoll (..),+     -- * Topic Routing (pgmq 1.11.0+)      -- ** Topic Management@@ -143,6 +158,7 @@     bindTopic,     changeVisibilityTimeout,     createPartitionedQueue,+    createPartitionedQueueWithPremake,     createQueue,     createUnloggedQueue,     deleteAllMessagesFromQueue,@@ -159,6 +175,12 @@     listTopicBindingsForQueue,     pop,     queueMetrics,+    readGrouped,+    readGroupedHead,+    readGroupedHeadWithPoll,+    readGroupedRoundRobin,+    readGroupedRoundRobinWithPoll,+    readGroupedWithPoll,     readMessage,     readWithPoll,     sendMessage,@@ -192,6 +214,8 @@     MessageQuery (..),     PopMessage (..),     QueueMetrics (..),+    ReadGrouped (..),+    ReadGroupedWithPoll (..),     ReadMessage (..),     ReadWithPollMessage (..),     SendMessage (..),

@@ -105,6 +105,7 @@     <*> D.column (D.nonNullable D.int8)     <*> D.column (D.nonNullable D.timestamptz)     <*> D.column (D.nonNullable D.int8) -- queue_visible_length (pgmq 1.5.0+)+    <*> D.column (D.nullable D.int8) -- default_partition_length (pgmq 1.13.0+)  -- | Decoder for topic binding records (pgmq 1.11.0+) -- Column order: pattern, queue_name, bound_at, compiled_regex

@@ -23,6 +23,7 @@     enableNotifyInsertEncoder,     readWithPollEncoder,     createPartitionedQueueEncoder,+    createPartitionedQueueWithPremakeEncoder,     -- FIFO encoders (pgmq 1.8.0+)     readGroupedEncoder,     readGroupedWithPollEncoder,@@ -214,8 +215,13 @@     <> (view #partitionInterval >$< E.param (E.nonNullable E.text))     <> (view #retentionInterval >$< E.param (E.nonNullable E.text)) +createPartitionedQueueWithPremakeEncoder :: E.Params (CreatePartitionedQueue, Int32)+createPartitionedQueueWithPremakeEncoder =+  (fst >$< createPartitionedQueueEncoder)+    <> (snd >$< E.param (E.nonNullable E.int4))+ -- | Encoder for ReadGrouped (pgmq 1.8.0+)--- Used for read_grouped and read_grouped_rr+-- Used for read_grouped, read_grouped_rr and read_grouped_head readGroupedEncoder :: E.Params ReadGrouped readGroupedEncoder =   (view #queueName >$< E.param (E.nonNullable queueNameValue))@@ -223,7 +229,7 @@     <> (view #qty >$< E.param (E.nonNullable E.int4))  -- | Encoder for ReadGroupedWithPoll (pgmq 1.8.0+)--- Used for read_grouped_with_poll and read_grouped_rr_with_poll+-- Used for read_grouped_with_poll, read_grouped_rr_with_poll and read_grouped_head_with_poll readGroupedWithPollEncoder :: E.Params ReadGroupedWithPoll readGroupedWithPollEncoder =   (view #queueName >$< E.param (E.nonNullable queueNameValue))

@@ -2,6 +2,7 @@   ( createQueue,     dropQueue,     createPartitionedQueue,+    createPartitionedQueueWithPremake,     createUnloggedQueue,     detachArchive,     enableNotifyInsert,@@ -38,6 +39,9 @@     -- Round-robin FIFO functions (pgmq 1.9.0+)     readGroupedRoundRobin,     readGroupedRoundRobinWithPoll,+    -- Grouped-head reads (pgmq 1.12.0+)+    readGroupedHead,+    readGroupedHeadWithPoll,     -- FIFO index functions (pgmq 1.8.0+)     createFifoIndex,     createFifoIndexesAll,@@ -198,6 +202,12 @@ createPartitionedQueue :: CreatePartitionedQueue -> Session () createPartitionedQueue q = statement q Stmt.createPartitionedQueue +-- | Create with an explicit number of partitions made ahead (pgmq 1.13.0+).+-- Counts below one are server errors. On 1.12 this operation is unsupported;+-- use 'createPartitionedQueue' for the server's default premake count.+createPartitionedQueueWithPremake :: CreatePartitionedQueue -> Int32 -> Session ()+createPartitionedQueueWithPremake q premake = statement (q, premake) Stmt.createPartitionedQueueWithPremake+ createUnloggedQueue :: QueueName -> Session () createUnloggedQueue q = statement q Stmt.createUnloggedQueue @@ -244,6 +254,19 @@ -- | Round-robin FIFO read with polling (pgmq 1.9.0+) readGroupedRoundRobinWithPoll :: ReadGroupedWithPoll -> Session (Vector Message) readGroupedRoundRobinWithPoll query = statement query Msg.readGroupedRoundRobinWithPoll++-- | Lease at most one absolute head per group (pgmq 1.12.0+).+-- @qty@ bounds the number of groups. An invisible lowest-ID message blocks+-- its group; lease expiry redelivers that same head. Delete or archive it to+-- advance the group. Messages without a group header share one implicit group.+-- A visibility lease does not guarantee exactly-once processing.+readGroupedHead :: ReadGrouped -> Session (Vector Message)+readGroupedHead query = statement query Msg.readGroupedHead++-- | Poll for 'readGroupedHead' results (pgmq 1.12.0+).+-- Occupies the database connection until work arrives or the poll times out.+readGroupedHeadWithPoll :: ReadGroupedWithPoll -> Session (Vector Message)+readGroupedHeadWithPoll query = statement query Msg.readGroupedHeadWithPoll  -- | Create FIFO index for a queue (pgmq 1.8.0+) createFifoIndex :: QueueName -> Session ()

@@ -26,6 +26,9 @@     -- Round-robin FIFO functions (pgmq 1.9.0+)     readGroupedRoundRobin,     readGroupedRoundRobinWithPoll,+    -- Grouped-head reads (pgmq 1.12.0+)+    readGroupedHead,+    readGroupedHeadWithPoll,     -- Topic send functions (pgmq 1.11.0+)     sendTopic,     sendTopicWithHeaders,@@ -310,6 +313,23 @@   where     sql = "select * from pgmq.read_grouped_rr_with_poll($1,$2,$3,$4,$5)"     decoder = D.rowVector messageDecoder++-- | Lease at most one absolute head per group (pgmq 1.12.0+).+-- @qty@ bounds the number of groups. An invisible lowest-ID message blocks+-- its group; lease expiry redelivers that same head. Delete or archive it to+-- advance the group. Messages without a group header share one implicit group.+-- A visibility lease does not guarantee exactly-once processing.+readGroupedHead :: Statement ReadGrouped (Vector Message)+readGroupedHead = preparable sql readGroupedEncoder (D.rowVector messageDecoder)+  where+    sql = "select * from pgmq.read_grouped_head($1,$2,$3)"++-- | Poll for 'readGroupedHead' results (pgmq 1.12.0+).+-- Occupies the database connection until work arrives or the poll times out.+readGroupedHeadWithPoll :: Statement ReadGroupedWithPoll (Vector Message)+readGroupedHeadWithPoll = preparable sql readGroupedWithPollEncoder (D.rowVector messageDecoder)+  where+    sql = "select * from pgmq.read_grouped_head_with_poll($1,$2,$3,$4,$5)"  -- | Send a message via topic routing (pgmq 1.11.0+) -- Returns the count of queues the message was delivered to.

@@ -2,6 +2,7 @@   ( createQueue,     dropQueue,     createPartitionedQueue,+    createPartitionedQueueWithPremake,     createUnloggedQueue,     detachArchive,     enableNotifyInsert,@@ -15,11 +16,12 @@   ) where +import Data.Int (Int32) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable) import Pgmq.Hasql.Decoders (notifyInsertThrottleDecoder)-import Pgmq.Hasql.Encoders (createPartitionedQueueEncoder, enableNotifyInsertEncoder, queueNameEncoder, updateNotifyInsertEncoder)+import Pgmq.Hasql.Encoders (createPartitionedQueueEncoder, createPartitionedQueueWithPremakeEncoder, enableNotifyInsertEncoder, queueNameEncoder, updateNotifyInsertEncoder) import Pgmq.Hasql.Statements.Types (CreatePartitionedQueue, EnableNotifyInsert, UpdateNotifyInsert) import Pgmq.Types (NotifyInsertThrottle, QueueName) @@ -41,6 +43,13 @@ createPartitionedQueue = preparable sql createPartitionedQueueEncoder D.noResult   where     sql = "select from pgmq.create_partitioned($1,$2,$3)"++-- | Create with an explicit premake count (pgmq 1.13.0+).+-- Counts below one are server errors; no fallback is attempted on older servers.+createPartitionedQueueWithPremake :: Statement (CreatePartitionedQueue, Int32) ()+createPartitionedQueueWithPremake = preparable sql createPartitionedQueueWithPremakeEncoder D.noResult+  where+    sql = "select from pgmq.create_partitioned($1,$2,$3,$4)"  -- | https://pgmq.github.io/pgmq/api/sql/functions/#create_unlogged createUnloggedQueue :: Statement QueueName ()

@@ -50,16 +50,26 @@       \where schemaname = 'pgmq' and indexname ~ '^q_.*_fifo_idx$'"     decoder = D.rowList (D.column (D.nonNullable D.text)) +-- | Metrics with a stable projection on pgmq 1.12 and 1.13.+-- The JSON record lookup preserves SQL NULL for the missing 1.12 attribute. -- | https://pgmq.github.io/pgmq/api/sql/functions/#metrics queueMetrics :: Statement QueueName QueueMetrics queueMetrics = preparable sql queueNameEncoder decoder   where-    sql = "select * from pgmq.metrics($1)"+    sql =+      "select m.queue_name, m.queue_length, m.newest_msg_age_sec, m.oldest_msg_age_sec, \+      \m.total_messages, m.scrape_time, m.queue_visible_length, \+      \(to_jsonb(m)->>'default_partition_length')::bigint \+      \from pgmq.metrics($1) as m"     decoder = D.singleRow queueMetricsDecoder  -- | https://pgmq.github.io/pgmq/api/sql/functions/#metrics_all allQueueMetrics :: Statement () [QueueMetrics] allQueueMetrics = preparable sql E.noParams decoder   where-    sql = "select * from pgmq.metrics_all()"+    sql =+      "select m.queue_name, m.queue_length, m.newest_msg_age_sec, m.oldest_msg_age_sec, \+      \m.total_messages, m.scrape_time, m.queue_visible_length, \+      \(to_jsonb(m)->>'default_partition_length')::bigint \+      \from pgmq.metrics_all() as m"     decoder = D.rowList queueMetricsDecoder

@@ -217,12 +217,16 @@     totalMessages :: !Int64,     scrapeTime :: !UTCTime,     -- | Count of messages available for reading (pgmq 1.5.0+)-    queueVisibleLength :: !Int64+    queueVisibleLength :: !Int64,+    -- | Planner estimate of rows in queue and archive default partitions (1.13+).+    -- 'Nothing' means unavailable on 1.12 or inapplicable for ordinary queues;+    -- it must not be interpreted as zero. Estimates can lag writes.+    defaultPartitionLength :: !(Maybe Int64)   }   deriving stock (Generic, Show)  -- | Parameters for FIFO grouped read (pgmq 1.8.0+)--- Used for both read_grouped and read_grouped_rr functions.+-- Used for read_grouped, read_grouped_rr and read_grouped_head functions. -- Note: conditional parameter was removed in pgmq 1.9.0 (commit 9e9c3dc) data ReadGrouped = ReadGrouped   { queueName :: !QueueName,

@@ -8,11 +8,14 @@ -- - read_grouped functions module AdvancedOpsSpec (tests) where -import Control.Concurrent (threadDelay)+import Control.Concurrent (forkFinally, killThread, newEmptyMVar, putMVar, takeMVar, threadDelay)+import Control.Exception (bracket, finally, throwIO) import Data.Aeson (object, (.=))+import Data.List (nub, sortOn) import Data.Time.Clock (addUTCTime, getCurrentTime) import Data.Vector qualified as V import EphemeralDb (TestFixture (..), withTestFixture)+import GHC.Clock (getMonotonicTimeNSec) import Hasql.Pool qualified as Pool import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types@@ -21,6 +24,7 @@     BatchSendMessageWithHeaders (..),     BatchVisibilityTimeoutAtQuery (..),     BatchVisibilityTimeoutQuery (..),+    MessageQuery (..),     PopMessage (..),     ReadGrouped (..),     ReadGroupedWithPoll (..),@@ -31,6 +35,7 @@   ) import Pgmq.Types (MessageBody (..), MessageHeaders (..), MessageId (..)) import Pgmq.Types qualified as PgmqTypes+import System.Timeout (timeout) import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertBool, assertEqual, testCase) import TestUtils (assertJust, assertSession, cleanupQueue)@@ -49,7 +54,8 @@       testReadWithPollEmpty p,       testCreateFifoIndex p,       testReadGrouped p,-      testReadGroupedRoundRobin p+      testReadGroupedRoundRobin p,+      groupedHeadTests p     ]  -- | Test pop with default qty (single message)@@ -377,3 +383,104 @@         Sessions.batchDeleteMessages           BatchMessageQuery {queueName = queueName, messageIds = msgIds}     cleanupQueue pool queueName++-- These assertions deliberately request more messages than there are groups.+-- Round-robin would fill six, whereas absolute-head reads can return only three.+groupedHeadTests :: Pool.Pool -> TestTree+groupedHeadTests p =+  testGroup+    "GroupedHead"+    [ testCase "absolute heads block successors and removal advances only that group" $+        withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+          ids <- seedHeads pool queueName+          let query = ReadGrouped queueName 60 6+          heads <- assertSession pool (Sessions.readGroupedHead query)+          assertEqual "absolute head IDs" [ids !! 0, ids !! 2, ids !! 4] (map PgmqTypes.messageId (sortOn PgmqTypes.messageId (V.toList heads)))+          assertEqual "head bodies" [headBody 1, headBody 3, headBody 5] (map PgmqTypes.body (sortOn PgmqTypes.messageId (V.toList heads)))+          assertEqual "three distinct groups" 3 (length (nub (map PgmqTypes.headers (V.toList heads))))+          blocked <- assertSession pool (Sessions.readGroupedHead query)+          assertEqual "invisible heads block all successors" V.empty blocked+          deleted <- assertSession pool (Sessions.deleteMessage (MessageQuery queueName (ids !! 0)))+          assertBool "deleted head" deleted+          advanced <- assertSession pool (Sessions.readGroupedHead query)+          assertEqual "only A advances after deletion" [ids !! 1] (map PgmqTypes.messageId (V.toList advanced))+          archived <- assertSession pool (Sessions.archiveMessage (MessageQuery queueName (ids !! 2)))+          assertBool "archived head" archived+          advancedAgain <- assertSession pool (Sessions.readGroupedHead query)+          assertEqual "only B advances after archive" [ids !! 3] (map PgmqTypes.messageId (V.toList advancedAgain)),+      testCase "lease expiry redelivers the same head with a higher read count" $+        withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+          _ <- seedHeads pool queueName+          let query = ReadGrouped queueName 60 1+          first <- assertSession pool (Sessions.readGroupedHead query)+          assertEqual "qty bounds groups" 1 (V.length first)+          let headMessage = V.head first+          past <- addUTCTime (-1) <$> getCurrentTime+          _ <- assertSession pool (Sessions.setVisibilityTimeoutAt (VisibilityTimeoutAtQuery queueName (PgmqTypes.messageId headMessage) past))+          again <- assertSession pool (Sessions.readGroupedHead (ReadGrouped queueName 60 6))+          let redelivered = V.filter ((== PgmqTypes.messageId headMessage) . PgmqTypes.messageId) again+          assertEqual "same ID redelivered" 1 (V.length redelivered)+          assertEqual "read count increases" (PgmqTypes.readCount headMessage + 1) (PgmqTypes.readCount (V.head redelivered)),+      testCase "missing headers form one implicit group" $+        withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+          assertSession pool (Sessions.createQueue queueName)+          assertSession pool (Sessions.createFifoIndex queueName)+          ids <- assertSession pool (Sessions.batchSendMessage (BatchSendMessage queueName [headBody 1, headBody 2] Nothing))+          heads <- assertSession pool (Sessions.readGroupedHead (ReadGrouped queueName 60 6))+          assertEqual "one implicit head" (take 1 ids) (map PgmqTypes.messageId (V.toList heads))+          blocked <- assertSession pool (Sessions.readGroupedHead (ReadGrouped queueName 60 6))+          assertEqual "implicit group blocks" V.empty blocked,+      testCase "poll returns available absolute heads promptly" $+        withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+          ids <- seedHeads pool queueName+          (elapsed, heads) <- timedPoll (assertSession pool (Sessions.readGroupedHeadWithPoll (ReadGroupedWithPoll queueName 60 6 5 50)))+          assertBool "ready work returns before poll deadline" (elapsed < 4)+          assertEqual "poll preserves head identity" [ids !! 0, ids !! 2, ids !! 4] (map PgmqTypes.messageId (sortOn PgmqTypes.messageId (V.toList heads))),+      testCase "empty poll waits then returns empty" $+        withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+          assertSession pool (Sessions.createQueue queueName)+          (elapsed, heads) <- timedPoll (assertSession pool (Sessions.readGroupedHeadWithPoll (ReadGroupedWithPoll queueName 60 6 1 50)))+          assertEqual "empty result" V.empty heads+          assertBool "waits about one second with scheduling margin" (elapsed >= 0.8 && elapsed < 4),+      testCase "poll sees a separately committed arrival" $+        withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+          assertSession pool (Sessions.createQueue queueName)+          -- The pool has three connections. The poll occupies one while this+          -- sender checks out another; forkFinally propagates sender failures.+          result <- newEmptyMVar+          bracket+            (forkFinally (threadDelay 300000 >> assertSession pool (Sessions.sendMessage (SendMessage queueName (headBody 1) Nothing))) (putMVar result))+            killThread+            ( \_ -> do+                (elapsed, heads) <- timedPoll (assertSession pool (Sessions.readGroupedHeadWithPoll (ReadGroupedWithPoll queueName 60 6 5 50)))+                sent <- assertJust =<< timeout 8000000 (takeMVar result)+                ident <- either throwIO pure sent+                assertEqual "arriving message returned" [ident] (map PgmqTypes.messageId (V.toList heads))+                assertBool "arrival precedes poll deadline" (elapsed < 4)+            )+    ]++headBody :: Int -> MessageBody+headBody n = MessageBody (object ["msg" .= n])++seedHeads :: Pool.Pool -> PgmqTypes.QueueName -> IO [MessageId]+seedHeads pool queueName = do+  assertSession pool (Sessions.createQueue queueName)+  assertSession pool (Sessions.createFifoIndex queueName)+  assertSession+    pool+    ( Sessions.batchSendMessageWithHeaders+        ( BatchSendMessageWithHeaders+            queueName+            (map headBody [1 .. 6])+            [MessageHeaders (object ["x-pgmq-group" .= group]) | group <- ["A", "A", "B", "B", "C", "C" :: String]]+            Nothing+        )+    )++timedPoll :: IO a -> IO (Double, a)+timedPoll action = do+  start <- getMonotonicTimeNSec+  result <- assertJust =<< timeout 8000000 action+  end <- getMonotonicTimeNSec+  pure (fromIntegral (end - start) / 1000000000, result)

@@ -16,9 +16,11 @@   ) where +import Control.Monad (filterM, when) import Data.List.NonEmpty (NonEmpty (..)) import Data.Text qualified as T-import Data.Word (Word32)+import Data.Text.IO qualified as TextIO+import Data.Word (Word64) import Database.PostgreSQL.Migrate   ( defaultRunOptions,     migrationPlan,@@ -32,8 +34,11 @@   ) 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 Pgmq.Types (QueueName, parseQueueName)+import System.Directory (doesFileExist)+import System.Environment (lookupEnv) import System.Random (randomRIO)  -- | Run an action with a temporary PostgreSQL database that has pgmq schema installed.@@ -48,12 +53,31 @@             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 db+  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-hasql/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 db+  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  -- | Run an action with a connection pool to a temporary PostgreSQL database -- The database will have the pgmq schema installed@@ -76,7 +100,8 @@ -- | Generate a random queue name for test isolation generateTestQueueName :: IO QueueName generateTestQueueName = do-  suffix <- randomRIO (10000 :: Word32, 99999)+  -- Property tests allocate hundreds of queues; a five-digit namespace collides.+  suffix <- randomRIO (0 :: Word64, maxBound)   case parseQueueName ("test_queue_" <> T.pack (show suffix)) of     Left err -> error $ "Failed to generate queue name: " <> show err     Right name -> pure name

@@ -14,6 +14,7 @@ import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KeyMap import Data.Scientific (fromFloatDigits)+import Data.Text qualified as Text import Data.Vector qualified as V import Hedgehog (Gen) import Hedgehog.Gen qualified as Gen@@ -86,7 +87,8 @@ genString = do   len <- Gen.int (Range.linear 0 100)   txt <- Gen.text (Range.singleton len) Gen.unicode-  pure $ String txt+  -- PostgreSQL jsonb cannot represent NUL, including while shrinking failures.+  pure $ String (Text.filter (/= '\0') txt)  genJsonArray :: Gen Value genJsonArray = do

@@ -8,6 +8,7 @@ import DecoderValidationSpec qualified import EphemeralDb (withPgmqDb) import MessageSpec qualified+import MetricsSpec qualified import MixedCaseRemediationSpec qualified import NotifyChannelSpec qualified import NotifyRaceSpec qualified@@ -16,38 +17,43 @@ import QueueSpec qualified import RoundTripSpec qualified import SchemaSpec qualified+import System.Environment (lookupEnv) import Test.Tasty (defaultMain, testGroup) import TopicSpec qualified+import UmbrellaExportsSpec ()  main :: IO () main = do+  stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"   -- Run tests with a shared temporary database   result <- withPgmqDb $ \pool db -> do     let tree =           testGroup             "pgmq-hasql"-            [ QueueSpec.tests pool,-              MessageSpec.tests pool,-              AdvancedOpsSpec.tests pool,-              NullSemanticsSpec.tests pool,-              NullBodySpec.tests pool,-              NotifyRaceSpec.tests pool,-              -- Both construct mixed-case pgmq.meta rows, which poison-              -- listQueues decoding for every concurrent test — so each runs-              -- on its own dedicated PostgreSQL instance, never the shared-              -- pool. They are separate instances because the remediation-              -- sweeps every mixed-case row in its database.-              AliasingSpec.tests,-              MixedCaseRemediationSpec.tests,-              -- Needs the Database handle: LISTEN/NOTIFY has no hasql API, so-              -- the round-trip test opens a raw libpq connection.-              NotifyChannelSpec.tests pool db,-              SchemaSpec.tests pool,-              RoundTripSpec.tests pool,-              DecoderValidationSpec.tests pool,-              AllFunctionsDecoderSpec.tests pool,-              TopicSpec.tests pool-            ]+            ( [ QueueSpec.tests pool,+                MessageSpec.tests pool,+                MetricsSpec.tests,+                AdvancedOpsSpec.tests pool,+                NullSemanticsSpec.tests pool,+                NullBodySpec.tests pool,+                -- Both construct mixed-case pgmq.meta rows, which poison+                -- listQueues decoding for every concurrent test — so each runs+                -- on its own dedicated PostgreSQL instance, never the shared+                -- pool. They are separate instances because the remediation+                -- sweeps every mixed-case row in its database.+                AliasingSpec.tests,+                MixedCaseRemediationSpec.tests,+                -- Needs the Database handle: LISTEN/NOTIFY has no hasql API, so+                -- the round-trip test opens a raw libpq connection.+                NotifyChannelSpec.tests pool db,+                SchemaSpec.tests pool,+                RoundTripSpec.tests pool,+                DecoderValidationSpec.tests pool,+                AllFunctionsDecoderSpec.tests pool,+                TopicSpec.tests pool+              ]+                <> [NotifyRaceSpec.tests pool | not stock]+            )     defaultMain tree   case result of     Left err -> error $ "Failed to start temp database: " <> show err

@@ -3,31 +3,40 @@ -- | Tests for queue metrics operations module MetricsSpec (tests) where +import Control.Exception (finally) import Data.Aeson (object, (.=))+import Data.Int (Int64) import Data.Text (Text)-import EphemeralDb (TestFixture (..), withTestFixture)+import EphemeralDb (TestFixture (..), withPgmqPool, withTestFixture)+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.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types (QueueMetrics (..)) import Pgmq.Hasql.Statements.Types qualified as Types import Pgmq.Types (MessageBody (..), QueueName, parseQueueName, queueNameToText)+import System.Environment (lookupEnv) import Test.Tasty (TestTree, testGroup)-import Test.Tasty.HUnit (assertBool, assertEqual, testCase, (@?=))+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase, (@?=)) import TestUtils (assertRight, assertSession, cleanupQueue)  -- | All metrics tests-tests :: Pool.Pool -> TestTree-tests p =+tests :: TestTree+tests =   testGroup     "Metrics"-    [ testQueueMetrics p,-      testQueueMetricsEmpty p,-      testAllQueueMetrics p+    [ testQueueMetrics,+      testQueueMetricsEmpty,+      testAllQueueMetrics,+      testUnloggedMetrics,+      testPartitionMetrics     ]  -- | Test queueMetrics returns correct metrics-testQueueMetrics :: Pool.Pool -> TestTree-testQueueMetrics p = testCase "queueMetrics returns queue statistics" $ do+testQueueMetrics :: TestTree+testQueueMetrics = testCase "queueMetrics returns queue statistics" $ withMetricsPool $ \p -> do   withTestFixture p $ \TestFixture {pool, queueName = qName} -> do     assertSession pool (Sessions.createQueue qName)     -- Send some messages@@ -38,6 +47,7 @@     -- Get metrics     metrics <- assertSession pool (Sessions.queueMetrics qName)     -- Verify metrics+    defaultPartitionLength metrics @?= Nothing     queueLength metrics @?= 3     totalMessages metrics @?= 3     queueVisibleLength metrics @?= 3@@ -45,20 +55,21 @@     cleanupQueue pool qName  -- | Test queueMetrics for empty queue-testQueueMetricsEmpty :: Pool.Pool -> TestTree-testQueueMetricsEmpty p = testCase "queueMetrics works for empty queue" $ do+testQueueMetricsEmpty :: TestTree+testQueueMetricsEmpty = testCase "queueMetrics works for empty queue" $ withMetricsPool $ \p -> do   withTestFixture p $ \TestFixture {pool, queueName = qName} -> do     assertSession pool (Sessions.createQueue qName)     -- Get metrics for empty queue     metrics <- assertSession pool (Sessions.queueMetrics qName)+    defaultPartitionLength metrics @?= Nothing     queueLength metrics @?= 0     totalMessages metrics @?= 0     queueVisibleLength metrics @?= 0     cleanupQueue pool qName  -- | Test allQueueMetrics returns all queues-testAllQueueMetrics :: Pool.Pool -> TestTree-testAllQueueMetrics p = testCase "allQueueMetrics returns metrics for all queues" $ do+testAllQueueMetrics :: TestTree+testAllQueueMetrics = testCase "allQueueMetrics returns metrics for all queues" $ withMetricsPool $ \p -> do   queueName1 <- assertRight $ parseQueueName "test_metrics_q1"   queueName2 <- assertRight $ parseQueueName "test_metrics_q2"   -- Create two queues@@ -76,7 +87,9 @@   -- Verify queue 1 has the correct message count   let q1Metrics = filter (\m -> metricsQueueName m == queueNameToText queueName1) allMetrics   case q1Metrics of-    [m] -> queueLength m @?= 1+    [m] -> do+      queueLength m @?= 1+      defaultPartitionLength m @?= Nothing     _ -> assertBool "Queue 1 should have metrics" False   -- Cleanup   cleanupQueue p queueName1@@ -94,3 +107,71 @@       Types.messageBody = MessageBody (object ["test" .= (1 :: Int)]),       Types.delay = Nothing     }++testUnloggedMetrics :: TestTree+testUnloggedMetrics = testCase "unlogged queues have no default-partition metric" $ withMetricsPool $ \p ->+  withTestFixture p $ \TestFixture {pool, queueName = qName} -> flip finally (cleanupQueue pool qName) $ do+    assertSession pool (Sessions.createUnloggedQueue qName)+    assertDefaultMetric pool qName Nothing++testPartitionMetrics :: TestTree+testPartitionMetrics = testCase "default-partition estimates include archived spills on both metrics APIs" $ withMetricsPool $ \p -> do+  available <-+    assertSession p $+      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 not available+    then+      if required+        then assertFailure "PGMQ_REQUIRE_PARTMAN=1 but pg_partman is not installed"+        else putStrLn "    SKIPPED: pg_partman is not installed"+    else withTestFixture p $ \TestFixture {pool, queueName = qName} -> flip finally (cleanupQueue pool qName) $ do+      stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+      let request = Types.CreatePartitionedQueue qName "10" "100"+          q = queueNameToText qName+          analyze = Session.script ("ANALYZE pgmq.q_" <> q <> "_default; ANALYZE pgmq.a_" <> q <> "_default")+      if stock+        then assertSession pool (Sessions.createPartitionedQueue request)+        else assertSession pool (Sessions.createPartitionedQueueWithPremake request 2)+      assertSession pool analyze+      assertDefaultMetric pool qName (if stock then Nothing else Just 0)+      -- Sending far beyond the premade partitions forces queue and archive spills.+      ids <-+        assertSession+          pool+          ( Sessions.batchSendMessage+              ( Types.BatchSendMessage+                  qName+                  [MessageBody (object ["n" .= n]) | n <- [1 .. 100 :: Int]]+                  Nothing+              )+          )+      let spilled = ids !! 99+      archived <- assertSession pool (Sessions.archiveMessage (Types.MessageQuery qName spilled))+      archived @?= True+      assertSession pool analyze+      (queueRows, archiveRows) <-+        assertSession pool $+          Session.statement () $+            preparable+              ("select (select count(*) from pgmq.q_" <> q <> "_default), (select count(*) from pgmq.a_" <> q <> "_default)")+              E.noParams+              (D.singleRow ((,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.int8)))+      assertBool "queue contains spilled messages" (queueRows > 0)+      archiveRows @?= 1+      assertDefaultMetric pool qName (if stock then Nothing else Just (queueRows + archiveRows))++assertDefaultMetric :: Pool.Pool -> QueueName -> Maybe Int64 -> IO ()+assertDefaultMetric pool qName expected = do+  single <- assertSession pool (Sessions.queueMetrics qName)+  defaultPartitionLength single @?= expected+  allMetrics <- assertSession pool Sessions.allQueueMetrics+  let matching = filter ((== queueNameToText qName) . metricsQueueName) allMetrics+  assertEqual "exactly one metrics_all row" 1 (length matching)+  map defaultPartitionLength matching @?= [expected]++-- metrics_all enumerates every queue, so unrelated concurrent queue drops can+-- invalidate its table lookup. Each case needs a stable database-wide fixture.+withMetricsPool :: (Pool.Pool -> IO ()) -> IO ()+withMetricsPool action = assertRight =<< withPgmqPool (\pool -> action pool `finally` Pool.release pool)

@@ -3,7 +3,12 @@ -- | Tests for queue management operations module QueueSpec (tests) where -import EphemeralDb (TestFixture (..), withTestFixture)+import Control.Exception (finally)+import Control.Monad (forM_)+import Data.Int (Int32)+import Data.List (isInfixOf)+import Data.Text qualified+import EphemeralDb (TestFixture (..), withPgmqPool, withTestFixture) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Pool qualified as Pool@@ -11,9 +16,10 @@ import Hasql.Statement (Statement, preparable) import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types qualified as StmtTypes-import Pgmq.Types (Queue (..), parseQueueName)+import Pgmq.Types (Queue (..), parseQueueName, queueNameToText)+import System.Environment (lookupEnv) import Test.Tasty (TestTree, testGroup)-import Test.Tasty.HUnit (assertBool, testCase, (@?=))+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=)) import TestUtils   ( assertRight,     assertSession,@@ -30,10 +36,8 @@       testDropNonExistentQueue p,       testListQueues p,       testCreateUnloggedQueue p,-      -- Note: the partitioned-queue tests need the pg_partman extension, which-      -- is not present in the ephemeral test environment; this one reports the-      -- skip rather than pretending to have run.-      testCreatePartitionedQueueIsReentrant p+      testCreatePartitionedQueueIsReentrant p,+      partitionCompatibilityTests p     ]  testCreateQueue :: Pool.Pool -> TestTree@@ -90,25 +94,33 @@ testCreatePartitionedQueueIsReentrant :: Pool.Pool -> TestTree testCreatePartitionedQueueIsReentrant p =   testCase "createPartitionedQueue is re-entrant (needs pg_partman)" $ do+    stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+    required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"     available <- assertSession p (Session.statement () pgPartmanAvailable)-    if not available-      then putStrLn "    SKIPPED: pg_partman is not available in this PostgreSQL installation"-      else do-        qName <- assertRight $ parseQueueName "test_partitioned_reentry"-        let request =-              StmtTypes.CreatePartitionedQueue-                { StmtTypes.queueName = qName,-                  StmtTypes.partitionInterval = "10000",-                  StmtTypes.retentionInterval = "100000"-                }-        assertSession p (Sessions.createPartitionedQueue request)-        assertSession p (Sessions.createPartitionedQueue request)-        cleanupQueue p qName+    if stock+      then putStrLn "    SKIPPED: local partition re-entry is a native-ledger contract"+      else+        if not available+          then+            if required+              then assertFailure "PGMQ_REQUIRE_PARTMAN=1 but pg_partman is not installed"+              else putStrLn "    SKIPPED: pg_partman is not installed in this PostgreSQL database"+          else do+            qName <- assertRight $ parseQueueName "test_partitioned_reentry"+            let request =+                  StmtTypes.CreatePartitionedQueue+                    { StmtTypes.queueName = qName,+                      StmtTypes.partitionInterval = "10000",+                      StmtTypes.retentionInterval = "100000"+                    }+            assertSession p (Sessions.createPartitionedQueue request)+            assertSession p (Sessions.createPartitionedQueue request)+            cleanupQueue p qName  pgPartmanAvailable :: Statement () Bool pgPartmanAvailable = preparable sql E.noParams decoder   where-    sql = "select exists (select 1 from pg_available_extensions where name = 'pg_partman')"+    sql = "select exists (select 1 from pg_extension where extname = 'pg_partman')"     decoder = D.singleRow (D.column (D.nonNullable D.bool))  testCreateUnloggedQueue :: Pool.Pool -> TestTree@@ -122,3 +134,80 @@   assertBool "Queue should exist" (not (null matchingQueues))   -- Cleanup   cleanupQueue p qName++partitionCompatibilityTests :: Pool.Pool -> TestTree+partitionCompatibilityTests p =+  testGroup+    "PartitionCompatibility"+    [ testCase "legacy creation keeps default premake on both versions" $+        withPartman p $+          withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+            assertSession pool (Sessions.createPartitionedQueue (StmtTypes.CreatePartitionedQueue queueName "10" "100"))+            counts <- premakeCounts pool (queueNameToText queueName)+            counts @?= [4, 4],+      testCase "explicit premake controls both parents or reports unsupported 1.12" $+        withPartman p $+          withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+            stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+            let request = StmtTypes.CreatePartitionedQueue queueName "10" "100"+            if stock+              then do+                result <- Pool.use pool (Sessions.createPartitionedQueueWithPremake request 2)+                case result of+                  Left err -> assertBool "undefined function SQLSTATE" ("42883" `isInfixOf` show err)+                  Right () -> assertFailure "1.12 must not silently discard explicit premake"+                assertNoPartitionQueue pool (queueNameToText queueName)+              else do+                assertSession pool (Sessions.createPartitionedQueueWithPremake request 2)+                counts <- premakeCounts pool (queueNameToText queueName)+                counts @?= [2, 2]+                assertSession pool (Sessions.createPartitionedQueueWithPremake request 2),+      testCase "zero and negative premake fail without creating objects" $+        forM_ [0, -1] $ \count -> do+          -- A failed Parse on the pinned Hasql leaves a prepared-statement+          -- cache entry. Isolate each unsupported-signature check so its first+          -- server error is observed instead of a later missing-prepared error.+          result <- withPgmqPool $ \isolated -> withPartman isolated $+            withTestFixture isolated $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do+              stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"+              creation <- Pool.use pool (Sessions.createPartitionedQueueWithPremake (StmtTypes.CreatePartitionedQueue queueName "10" "100") count)+              case creation of+                Left err ->+                  assertBool+                    ("documented database error: " <> show err)+                    ((if stock then "42883" else "premake must be at least 1") `isInfixOf` show err)+                Right () -> assertFailure "invalid premake unexpectedly succeeded"+              assertNoPartitionQueue pool (queueNameToText queueName)+          assertRight result+    ]++withPartman :: Pool.Pool -> IO () -> IO ()+withPartman pool action = do+  available <- assertSession pool (Session.statement () pgPartmanAvailable)+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"+  if available+    then action+    else+      if required+        then assertFailure "PGMQ_REQUIRE_PARTMAN=1 but pg_partman is not installed"+        else putStrLn "    SKIPPED: pg_partman is not installed"++premakeCounts :: Pool.Pool -> Data.Text.Text -> IO [Int32]+premakeCounts pool queue =+  assertSession pool $+    Session.statement queue $+      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)))++assertNoPartitionQueue :: Pool.Pool -> Data.Text.Text -> IO ()+assertNoPartitionQueue pool queue = do+  absent <-+    assertSession pool $+      Session.statement queue $+        preparable+          "select not exists (select from pgmq.meta where queue_name = $1) and to_regclass('pgmq.q_' || $1) is null and to_regclass('pgmq.a_' || $1) is null and not exists (select from partman.part_config where parent_table in ('pgmq.q_' || $1, 'pgmq.a_' || $1))"+          (E.param (E.nonNullable E.text))+          (D.singleRow (D.column (D.nonNullable D.bool)))+  assertBool "failed creation leaves no metadata, tables or parent registrations" absent

@@ -0,0 +1,27 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeOperators #-}++-- Compile witnesses: this module intentionally uses only the public umbrella.+module UmbrellaExportsSpec (grouped, polling, creation, arguments, metric) where++import Data.Int (Int32, Int64)+import Data.Vector (Vector)+import Hasql.Session (Session)+import Pgmq++grouped :: ReadGrouped -> [Session (Vector Message)]+grouped args = [readGrouped args, readGroupedRoundRobin args, readGroupedHead args]++polling :: ReadGroupedWithPoll -> [Session (Vector Message)]+polling args = [readGroupedWithPoll args, readGroupedRoundRobinWithPoll args, readGroupedHeadWithPoll args]++creation :: CreatePartitionedQueue -> Int32 -> [Session ()]+creation args n = [createPartitionedQueue args, createPartitionedQueueWithPremake args n]++arguments :: QueueName -> (ReadGrouped, ReadGroupedWithPoll, CreatePartitionedQueue)+arguments q = (ReadGrouped q 30 10, ReadGroupedWithPoll q 30 10 5 100, CreatePartitionedQueue q "100" "1000")++metric :: QueueMetrics -> Maybe Int64+metric = defaultPartitionLength

@@ -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));+$$;