diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/pgmq-hasql.cabal b/pgmq-hasql.cabal --- a/pgmq-hasql.cabal +++ b/pgmq-hasql.cabal @@ -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, diff --git a/src/Pgmq.hs b/src/Pgmq.hs --- a/src/Pgmq.hs +++ b/src/Pgmq.hs @@ -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 (..), diff --git a/src/Pgmq/Hasql/Decoders.hs b/src/Pgmq/Hasql/Decoders.hs --- a/src/Pgmq/Hasql/Decoders.hs +++ b/src/Pgmq/Hasql/Decoders.hs @@ -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 diff --git a/src/Pgmq/Hasql/Encoders.hs b/src/Pgmq/Hasql/Encoders.hs --- a/src/Pgmq/Hasql/Encoders.hs +++ b/src/Pgmq/Hasql/Encoders.hs @@ -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)) diff --git a/src/Pgmq/Hasql/Sessions.hs b/src/Pgmq/Hasql/Sessions.hs --- a/src/Pgmq/Hasql/Sessions.hs +++ b/src/Pgmq/Hasql/Sessions.hs @@ -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 () diff --git a/src/Pgmq/Hasql/Statements/Message.hs b/src/Pgmq/Hasql/Statements/Message.hs --- a/src/Pgmq/Hasql/Statements/Message.hs +++ b/src/Pgmq/Hasql/Statements/Message.hs @@ -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. diff --git a/src/Pgmq/Hasql/Statements/QueueManagement.hs b/src/Pgmq/Hasql/Statements/QueueManagement.hs --- a/src/Pgmq/Hasql/Statements/QueueManagement.hs +++ b/src/Pgmq/Hasql/Statements/QueueManagement.hs @@ -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 () diff --git a/src/Pgmq/Hasql/Statements/QueueObservability.hs b/src/Pgmq/Hasql/Statements/QueueObservability.hs --- a/src/Pgmq/Hasql/Statements/QueueObservability.hs +++ b/src/Pgmq/Hasql/Statements/QueueObservability.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements/Types.hs b/src/Pgmq/Hasql/Statements/Types.hs --- a/src/Pgmq/Hasql/Statements/Types.hs +++ b/src/Pgmq/Hasql/Statements/Types.hs @@ -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, diff --git a/test/AdvancedOpsSpec.hs b/test/AdvancedOpsSpec.hs --- a/test/AdvancedOpsSpec.hs +++ b/test/AdvancedOpsSpec.hs @@ -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) diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs --- a/test/EphemeralDb.hs +++ b/test/EphemeralDb.hs @@ -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 diff --git a/test/Generators.hs b/test/Generators.hs --- a/test/Generators.hs +++ b/test/Generators.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 diff --git a/test/MetricsSpec.hs b/test/MetricsSpec.hs --- a/test/MetricsSpec.hs +++ b/test/MetricsSpec.hs @@ -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) diff --git a/test/QueueSpec.hs b/test/QueueSpec.hs --- a/test/QueueSpec.hs +++ b/test/QueueSpec.hs @@ -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 diff --git a/test/UmbrellaExportsSpec.hs b/test/UmbrellaExportsSpec.hs new file mode 100644 --- /dev/null +++ b/test/UmbrellaExportsSpec.hs @@ -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 diff --git a/test/fixtures/pgmq-1.12.0.sql b/test/fixtures/pgmq-1.12.0.sql new file mode 100644 --- /dev/null +++ b/test/fixtures/pgmq-1.12.0.sql @@ -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) + (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)); +$$;