diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Revision history for pgmq-config +## 0.6.0.0 -- 2026-09-10 + +Breaking: `PartitionConfig` adds `premake :: Maybe Int32`. Supply `Nothing` for the existing +server default (4), or `Just n` for an explicit count of at least 1 on PGMQ 1.13+. +Both pool and effect adapters forward the choice through the shared internal `ReconcileOps`. +Premake is creation-only: existing queue settings are neither changed nor checked for drift. +`ReconcileOps` remains internal; no external backend migration is required. + ## 0.5.0.0 -- 2026-08-06 The reconciler told three lies about what it had done, and could be taken down at startup diff --git a/pgmq-config.cabal b/pgmq-config.cabal --- a/pgmq-config.cabal +++ b/pgmq-config.cabal @@ -1,6 +1,6 @@ -cabal-version: 3.4 -name: pgmq-config -version: 0.5.0.0 +cabal-version: 3.4 +name: pgmq-config +version: 0.6.0.0 synopsis: Declarative queue configuration for PGMQ (PostgreSQL Message Queue) @@ -19,34 +19,41 @@ reported as drift rather than repaired, since converting a queue's type would mean destroying its messages. -homepage: https://github.com/shinzui/pgmq-hs -license: MIT -license-file: LICENSE -author: Nadeem Bitar -maintainer: Nadeem Bitar -category: Database -build-type: Simple +homepage: https://github.com/shinzui/pgmq-hs +license: MIT +license-file: LICENSE +author: Nadeem Bitar +maintainer: Nadeem Bitar +category: Database +build-type: Simple extra-doc-files: CHANGELOG.md +extra-source-files: test/fixtures/pgmq-1.12.0.sql flag effectful description: Enable effectful integration - default: True - manual: False + default: True + manual: False common warnings ghc-options: - -Wall -Wcompat -Widentities -Wincomplete-uni-patterns - -Wincomplete-record-updates -Wredundant-constraints - -fhide-source-paths -Wmissing-export-lists -Wpartial-fields + -Wall + -Wcompat + -Widentities + -Wincomplete-uni-patterns + -Wincomplete-record-updates + -Wredundant-constraints + -fhide-source-paths + -Wmissing-export-lists + -Wpartial-fields -Wmissing-deriving-strategies library - import: warnings + import: warnings exposed-modules: Pgmq.Config Pgmq.Config.Types - other-modules: Pgmq.Config.Reconcile + other-modules: Pgmq.Config.Reconcile default-extensions: DeriveGeneric DuplicateRecordFields @@ -57,37 +64,41 @@ OverloadedStrings build-depends: - , base >=4.18 && <5 - , containers >=0.6 && <0.8 - , generic-lens ^>=2.2 || ^>=2.3 - , hasql ^>=1.10 - , hasql-pool ^>=1.4 - , lens ^>=5.3 - , pgmq-core >=0.5 && <0.6 - , pgmq-hasql >=0.5 && <0.6 - , text ^>=2.1 + base >=4.18 && <5, + containers >=0.6 && <0.8, + generic-lens ^>=2.2 || ^>=2.3, + hasql ^>=1.10, + hasql-pool ^>=1.4, + lens ^>=5.3, + pgmq-core >=0.6 && <0.7, + pgmq-hasql >=0.6 && <0.7, + text ^>=2.1, if flag(effectful) exposed-modules: Pgmq.Config.Effectful build-depends: - , effectful-core ^>=2.5 || ^>=2.6 - , pgmq-effectful >=0.5 && <0.6 - - hs-source-dirs: src - default-language: GHC2024 + effectful-core ^>=2.5 || ^>=2.6, + pgmq-effectful >=0.6 && <0.7, + hs-source-dirs: src + default-language: GHC2024 test-suite pgmq-config-test - import: warnings - default-language: GHC2024 - type: exitcode-stdio-1.0 - hs-source-dirs: test - main-is: Main.hs - ghc-options: -threaded -rtsopts -with-rtsopts=-N + import: warnings + default-language: GHC2024 + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Main.hs + ghc-options: + -threaded + -rtsopts + -with-rtsopts=-N + other-modules: ConfigSpec EphemeralDb ForeignQueueSpec NotifyCrashSpec + PartitionSpec default-extensions: DeriveGeneric @@ -99,31 +110,32 @@ OverloadedStrings build-depends: - , aeson ^>=2.2 - , base >=4.18 && <5 - , bytestring >=0.11 && <0.13 - , ephemeral-pg >=0.2.1 - , generic-lens ^>=2.2 || ^>=2.3 - , hasql - , hasql-pool ^>=1.4 - , lens ^>=5.3 - , pg-migrate - , pgmq-config - , pgmq-core - , pgmq-hasql - , pgmq-migration - , postgresql-libpq >=0.10.1 && <0.12 - , random ^>=1.2 - , tasty ^>=1.5 - , tasty-hunit ^>=0.10 - , text - , time ^>=1.14 + aeson ^>=2.2, + base >=4.18 && <5, + bytestring >=0.11 && <0.13, + directory, + ephemeral-pg >=0.2.1, + generic-lens ^>=2.2 || ^>=2.3, + hasql, + hasql-pool ^>=1.4, + lens ^>=5.3, + pg-migrate, + pgmq-config >=0.6 && <0.7, + pgmq-core >=0.6 && <0.7, + pgmq-hasql >=0.6 && <0.7, + pgmq-migration >=0.6 && <0.7, + postgresql-libpq >=0.10.1 && <0.12, + random ^>=1.2, + tasty ^>=1.5, + tasty-hunit ^>=0.10, + text, + time ^>=1.14, -- ForeignQueueSpec asserts that both reconciler backends use the lenient -- queue listing. The effect-backed half only exists when the library's -- effectful flag is on, so it is compiled in behind the same flag. if flag(effectful) - cpp-options: -DPGMQ_EFFECTFUL + cpp-options: -DPGMQ_EFFECTFUL build-depends: - , effectful-core ^>=2.5 || ^>=2.6 - , pgmq-effectful >=0.5 && <0.6 + effectful-core ^>=2.5 || ^>=2.6, + pgmq-effectful >=0.6 && <0.7, diff --git a/src/Pgmq/Config.hs b/src/Pgmq/Config.hs --- a/src/Pgmq/Config.hs +++ b/src/Pgmq/Config.hs @@ -58,6 +58,7 @@ createQueue = Sessions.createQueue, createUnloggedQueue = Sessions.createUnloggedQueue, createPartitionedQueue = Sessions.createPartitionedQueue, + createPartitionedQueueWithPremake = Sessions.createPartitionedQueueWithPremake, enableNotifyInsert = Sessions.enableNotifyInsert, createFifoIndex = Sessions.createFifoIndex, bindTopic = Sessions.bindTopic, @@ -94,7 +95,7 @@ -- Changing a queue's type means dropping and recreating it, destroying every -- message it holds; that is an operator's decision, not a startup task. -- --- * A partitioned queue's partition interval and retention interval are never +-- * A partitioned queue's partition interval, retention interval and premake are never -- compared, because @pgmq.list_queues()@ does not report them. Only the -- three-way shape is checked. -- diff --git a/src/Pgmq/Config/Effectful.hs b/src/Pgmq/Config/Effectful.hs --- a/src/Pgmq/Config/Effectful.hs +++ b/src/Pgmq/Config/Effectful.hs @@ -29,6 +29,7 @@ createQueue = Eff.createQueue, createUnloggedQueue = Eff.createUnloggedQueue, createPartitionedQueue = Eff.createPartitionedQueue, + createPartitionedQueueWithPremake = Eff.createPartitionedQueueWithPremake, enableNotifyInsert = Eff.enableNotifyInsert, createFifoIndex = Eff.createFifoIndex, bindTopic = Eff.bindTopic, diff --git a/src/Pgmq/Config/Reconcile.hs b/src/Pgmq/Config/Reconcile.hs --- a/src/Pgmq/Config/Reconcile.hs +++ b/src/Pgmq/Config/Reconcile.hs @@ -45,6 +45,7 @@ createQueue :: QueueName -> m (), createUnloggedQueue :: QueueName -> m (), createPartitionedQueue :: StmtTypes.CreatePartitionedQueue -> m (), + createPartitionedQueueWithPremake :: StmtTypes.CreatePartitionedQueue -> Int32 -> m (), enableNotifyInsert :: StmtTypes.EnableNotifyInsert -> m (), createFifoIndex :: QueueName -> m (), bindTopic :: StmtTypes.BindTopic -> m (), @@ -128,12 +129,15 @@ UnloggedQueue -> (ops ^. #createUnloggedQueue) qn PartitionedQueue pc -> - (ops ^. #createPartitionedQueue) - StmtTypes.CreatePartitionedQueue - { queueName = qn, - partitionInterval = pc ^. #partitionInterval, - retentionInterval = pc ^. #retentionInterval - } + let query = + StmtTypes.CreatePartitionedQueue + { queueName = qn, + partitionInterval = pc ^. #partitionInterval, + retentionInterval = pc ^. #retentionInterval + } + in case pc ^. #premake of + Nothing -> (ops ^. #createPartitionedQueue) query + Just n -> (ops ^. #createPartitionedQueueWithPremake) query n pure [CreatedQueue qn declaredType] -- Notification. A missing row is enabled; a row whose interval already @@ -180,7 +184,7 @@ pure (queueAction ++ notifyAction ++ fifoAction ++ bindingActions) -- | The shape a declared config asks for, reduced to what @pgmq.list_queues()@ --- can actually report. Partition interval and retention are deliberately +-- can actually report. Partition interval, retention and premake are deliberately -- dropped here: the listing does not expose them, so they are not drift-checked. declaredShape :: QueueType -> ObservedQueueType declaredShape StandardQueue = ObservedStandard diff --git a/src/Pgmq/Config/Types.hs b/src/Pgmq/Config/Types.hs --- a/src/Pgmq/Config/Types.hs +++ b/src/Pgmq/Config/Types.hs @@ -59,7 +59,11 @@ -- | Configuration for a partitioned queue. data PartitionConfig = PartitionConfig { partitionInterval :: !Text, - retentionInterval :: !Text + retentionInterval :: !Text, + -- | Creation only: Nothing uses the server default via the legacy call; + -- Just requires PGMQ 1.13+ and a count of at least one. Existing queues + -- are neither checked nor reconfigured for this setting. + premake :: !(Maybe Int32) } deriving stock (Generic, Show) diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs --- a/test/EphemeralDb.hs +++ b/test/EphemeralDb.hs @@ -10,7 +10,9 @@ ) where +import Control.Monad (filterM, when) import Data.List.NonEmpty (NonEmpty (..)) +import Data.Text.IO qualified as TextIO import Database.PostgreSQL.Migrate ( defaultRunOptions, migrationPlan, @@ -23,7 +25,10 @@ ) import Hasql.Pool qualified as Pool import Hasql.Pool.Config qualified as PoolConfig +import Hasql.Session qualified as Session import Pgmq.Migration qualified as Migration +import System.Directory (doesFileExist) +import System.Environment (lookupEnv) -- | Run an action with a temporary PostgreSQL database that has pgmq schema installed withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a) @@ -35,9 +40,28 @@ PoolConfig.staticConnectionSettings connSettings ] pool <- Pool.acquire poolConfig - component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations - plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| [])) - installResult <- runMigrationPlan defaultRunOptions connSettings plan - case installResult of - Left migrationErr -> error $ "Migration failed: " <> show migrationErr - Right _ -> action pool + version <- lookupEnv "PGMQ_TEST_SCHEMA_VERSION" + case version of + Just "1.12.0" -> do + paths <- filterM doesFileExist ["test/fixtures/pgmq-1.12.0.sql", "pgmq-config/test/fixtures/pgmq-1.12.0.sql"] + path <- case paths of + candidate : _ -> pure candidate + [] -> error "Missing packaged PGMQ 1.12.0 test fixture" + sql <- TextIO.readFile path + Pool.use pool (Session.script sql) >>= either (error . show) pure + Nothing -> installNative connSettings + Just "1.13.0" -> installNative connSettings + Just invalid -> error ("Invalid PGMQ_TEST_SCHEMA_VERSION: " <> invalid) + required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN" + -- Required runs fail for absent or unusable pg_partman, not just missing metadata. + partman <- Pool.use pool (Session.script "CREATE SCHEMA IF NOT EXISTS partman; CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman") + case partman of + Left err -> when required (error ("PGMQ_REQUIRE_PARTMAN=1: " <> show err)) + Right () -> pure () + action pool + where + installNative connSettings = do + component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations + plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| [])) + installResult <- runMigrationPlan defaultRunOptions connSettings plan + either (error . ("Migration failed: " <>) . show) (const (pure ())) installResult diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -6,6 +6,7 @@ import EphemeralDb (withPgmqDb) import ForeignQueueSpec qualified import NotifyCrashSpec qualified +import PartitionSpec qualified import Test.Tasty (defaultMain, testGroup) main :: IO () @@ -15,6 +16,7 @@ testGroup "pgmq-config" [ ConfigSpec.tests pool, + PartitionSpec.tests, -- NotifyCrashSpec manages its own PostgreSQL instance: it crashes -- the server, which the shared pool above could not survive. NotifyCrashSpec.tests, diff --git a/test/NotifyCrashSpec.hs b/test/NotifyCrashSpec.hs --- a/test/NotifyCrashSpec.hs +++ b/test/NotifyCrashSpec.hs @@ -144,7 +144,8 @@ bracket (acquirePool db1) Pool.release $ \pool -> do throttlesAfterCrash <- listThrottleNames pool triggers <- runSession pool (Session.statement (queueTableName qn) insertTriggerCount) - metrics <- runSession pool (Sessions.queueMetrics qn) + -- Keep the crash regression independent of the expanding metrics record. + queueLength <- runSession pool (Session.statement (queueNameToText qn) queueLengthStatement) let channel = notifyChannelName qn notified <- withListener db1 channel $ \conn -> do @@ -160,7 +161,7 @@ obsChannel = channel, obsThrottlesAfterCrash = throttlesAfterCrash, obsTriggersAfterCrash = triggers, - obsQueueLengthAfterCrash = metrics ^. #queueLength, + obsQueueLengthAfterCrash = queueLength, obsNotifyAfterCrash = LibPQ.notifyRelname <$> notified, obsThrottlesAfterReconcile = throttlesAfterReconcile } @@ -235,6 +236,13 @@ queueTableName qn = "q_" <> T.toLower (queueNameToText qn) -- | How many insert-notification triggers exist on the given @pgmq@ table. +queueLengthStatement :: Statement Text Int64 +queueLengthStatement = + preparable + "SELECT queue_length FROM pgmq.metrics($1)" + (E.param (E.nonNullable E.text)) + (D.singleRow (D.column (D.nonNullable D.int8))) + insertTriggerCount :: Statement Text Int64 insertTriggerCount = preparable sql encoder decoder where diff --git a/test/PartitionSpec.hs b/test/PartitionSpec.hs new file mode 100644 --- /dev/null +++ b/test/PartitionSpec.hs @@ -0,0 +1,111 @@ +{-# LANGUAGE CPP #-} + +module PartitionSpec (tests) where + +import Control.Exception (finally) +import Control.Monad (forM_) +import Data.Int (Int32) +import Data.List (isInfixOf) +import EphemeralDb (withPgmqDb) +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Pool qualified as Pool +import Hasql.Session qualified as Session +import Hasql.Statement (preparable) +import Pgmq.Config +import Pgmq.Types (QueueName, parseQueueName, queueNameToText) +import System.Environment (lookupEnv) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=)) +#ifdef PGMQ_EFFECTFUL +import Effectful (runEff) +import Effectful.Error.Static (runError) +import Pgmq.Config.Effectful (ensureQueuesReportEff) +import Pgmq.Effectful (PgmqRuntimeError, runPgmq) +#endif + +tests :: TestTree +tests = testGroup "Partition compatibility" [backendTests backend | backend <- backends] + +backends :: [Bool] +#ifdef PGMQ_EFFECTFUL +backends = [False, True] +#else +backends = [False] +#endif + +reconcile :: Bool -> Pool.Pool -> [QueueConfig] -> IO (Either String [ReconcileAction]) +#ifdef PGMQ_EFFECTFUL +reconcile True pool configs = either (Left . show) Right <$> runEff (runError @PgmqRuntimeError (runPgmq pool (ensureQueuesReportEff configs))) +#endif +reconcile _ pool configs = either (Left . show) Right <$> Pool.use pool (ensureQueuesReport configs) + +backendTests :: Bool -> TestTree +backendTests effectful = + testGroup + (if effectful then "effectful" else "direct") + $ [ testCase ("creation and skip " <> show requested) $ isolated $ \pool -> withPartman pool $ do + stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION" + qn <- right (parseQueueName "partition_config") + let config n = partitionedQueue qn (PartitionConfig "10" "100" n) + result <- reconcile effectful pool [config requested] + if stock && requested /= Nothing + then case result of + Left err -> assertBool "explicit premake requires 1.13" ("42883" `isInfixOf` err) + Right _ -> assertFailure "unsupported premake succeeded" + else do + actions <- right result + case actions of + [CreatedQueue q _] -> q @?= qn + other -> assertFailure ("expected creation: " <> show other) + let expected = maybe 4 id requested + counts pool qn >>= (@?= [expected, expected]) + -- Turn every creation call into an error, proving the existing-queue + -- path makes no call even when a different (invalid) count is requested. + let signature = "queue_name text, partition_interval text DEFAULT '10000', retention_interval text DEFAULT '100000'" <> if stock then "" else ", premake integer DEFAULT 4" + session pool (Session.script ("CREATE OR REPLACE FUNCTION pgmq.create_partitioned(" <> signature <> ") RETURNS void LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'unexpected partition creation'; END $$")) + skipped <- right =<< reconcile effectful pool [config (Just 0)] + case skipped of + [SkippedQueue q] -> q @?= qn + other -> assertFailure ("expected skip: " <> show other) + counts pool qn >>= (@?= [expected, expected]) + | requested <- [Nothing, Just 2] + ] + ++ [ testCase "zero and negative counts fail on new queues" $ forM_ [0, -1] $ \n -> isolated $ \pool -> withPartman pool $ do + stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION" + qn <- right (parseQueueName "invalid_partition_config") + result <- reconcile effectful pool [partitionedQueue qn (PartitionConfig "10" "100" (Just n))] + case result of + Left err -> assertBool "server error retained" ((if stock then "42883" else "premake must be at least 1") `isInfixOf` err) + Right _ -> assertFailure "invalid premake succeeded" + ] + +counts :: Pool.Pool -> QueueName -> IO [Int32] +counts pool qn = + session pool $ + Session.statement (queueNameToText qn) $ + preparable + "select premake from partman.part_config where parent_table in ('pgmq.q_' || $1, 'pgmq.a_' || $1) order by parent_table" + (E.param (E.nonNullable E.text)) + (D.rowList (D.column (D.nonNullable D.int4))) + +session :: Pool.Pool -> Session.Session a -> IO a +session pool action = right =<< Pool.use pool action + +right :: (Show e) => Either e a -> IO a +right = either (\err -> assertFailure (show err) >> fail "unreachable") pure + +isolated :: (Pool.Pool -> IO ()) -> IO () +isolated action = right =<< withPgmqDb (\pool -> action pool `finally` Pool.release pool) + +withPartman :: Pool.Pool -> IO () -> IO () +withPartman pool action = do + available <- + session pool $ + Session.statement () $ + preparable + "select exists (select from pg_extension where extname = 'pg_partman')" + E.noParams + (D.singleRow (D.column (D.nonNullable D.bool))) + required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN" + if available then action else if required then assertFailure "pg_partman required" else putStrLn "SKIPPED: pg_partman is unavailable" 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)); +$$;