pgmq-hasql 0.1.0.0 → 0.1.1.0

raw patch · 13 files changed

+945/−14 lines, 13 filesdep ~pgmq-corePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: pgmq-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +11/−0
• pgmq-hasql.cabal +7/−5
• src/Pgmq.hs +78/−0
• src/Pgmq/Hasql/Decoders.hs +53/−1
• src/Pgmq/Hasql/Encoders.hs +92/−0
• src/Pgmq/Hasql/Sessions.hs +106/−1
• src/Pgmq/Hasql/Statements.hs +2/−0
• src/Pgmq/Hasql/Statements/Message.hs +65/−2
• src/Pgmq/Hasql/Statements/QueueManagement.hs +20/−3
• src/Pgmq/Hasql/Statements/TopicManagement.hs +78/−0
• src/Pgmq/Hasql/Statements/Types.hs +83/−1
• test/Main.hs +3/−1
• test/TopicSpec.hs +347/−0

@@ -1,5 +1,16 @@ # Revision history for pgmq-hasql +## 0.1.1.0 -- 2026-02-23++### New Features++#### pgmq 1.11.0+ Support++* Topic management: `bindTopic`, `unbindTopic`, `validateRoutingKey`, `validateTopicPattern`, `testRouting`, `listTopicBindings`, `listTopicBindingsForQueue`+* Topic sending: `sendTopic`, `sendTopicWithHeaders`, `batchSendTopic`, `batchSendTopicForLater`, `batchSendTopicWithHeaders`, `batchSendTopicWithHeadersForLater`+* Notification management: `listNotifyInsertThrottles`, `updateNotifyInsert`+* New parameter types: `BindTopic`, `UnbindTopic`, `SendTopic`, `SendTopicWithHeaders`, `BatchSendTopic`, `BatchSendTopicForLater`, `BatchSendTopicWithHeaders`, `BatchSendTopicWithHeadersForLater`, `UpdateNotifyInsert`+ ## 0.1.0.0 -- 2026-02-21  ### New Features

@@ -1,19 +1,19 @@ cabal-version:   3.4 name:            pgmq-hasql-version:         0.1.0.0+version:         0.1.1.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,   visibility timeout, archival, FIFO reads, and more. Supports-  pgmq 1.5.0 through 1.10.0 features including message headers,-  conditional reads, notifications, and round-robin FIFO.+  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/topagentnetwork/pgmq-hs license:         MIT license-file:    LICENSE author:          Nadeem Bitar-maintainer:      nadeem@topagentnetwork.com+maintainer:      Nadeem Bitar category:        Database build-type:      Simple extra-doc-files: CHANGELOG.md@@ -38,6 +38,7 @@     Pgmq.Hasql.Statements.Message     Pgmq.Hasql.Statements.QueueManagement     Pgmq.Hasql.Statements.QueueObservability+    Pgmq.Hasql.Statements.TopicManagement     Pgmq.Hasql.Statements.Types    default-extensions:@@ -56,7 +57,7 @@     , hasql              ^>=1.10     , hasql-transaction  ^>=1.2     , lens               ^>=5.3-    , pgmq-core          >=0.1   && <0.2+    , pgmq-core          >=0.1.1 && <0.2     , template-haskell   >=2.20  && <3     , text               ^>=2.1     , time               ^>=1.14@@ -84,6 +85,7 @@     RoundTripSpec     SchemaSpec     TestUtils+    TopicSpec    default-extensions:     ImportQualifiedPost

@@ -39,6 +39,29 @@     queueMetrics,     allQueueMetrics, +    -- * Topic Routing (pgmq 1.11.0+)++    -- ** Topic Management+    bindTopic,+    unbindTopic,+    validateRoutingKey,+    validateTopicPattern,+    testRouting,+    listTopicBindings,+    listTopicBindingsForQueue,++    -- ** Topic Sending+    sendTopic,+    sendTopicWithHeaders,+    batchSendTopic,+    batchSendTopicForLater,+    batchSendTopicWithHeaders,+    batchSendTopicWithHeadersForLater,++    -- ** Notification Management+    listNotifyInsertThrottles,+    updateNotifyInsert,+     -- * Types     MessageBody (..),     MessageHeaders (..),@@ -71,6 +94,27 @@     CreatePartitionedQueue (..),     QueueMetrics (..), +    -- ** Topic types (pgmq 1.11.0+)+    RoutingKey,+    parseRoutingKey,+    routingKeyToText,+    TopicPattern,+    parseTopicPattern,+    topicPatternToText,+    TopicBinding (..),+    RoutingMatch (..),+    TopicSendResult (..),+    NotifyInsertThrottle (..),+    BindTopic (..),+    UnbindTopic (..),+    SendTopic (..),+    SendTopicWithHeaders (..),+    BatchSendTopic (..),+    BatchSendTopicForLater (..),+    BatchSendTopicWithHeaders (..),+    BatchSendTopicWithHeadersForLater (..),+    UpdateNotifyInsert (..),+     -- * Queue Name Utilities     parseQueueName,     queueNameToText,@@ -87,7 +131,12 @@     batchSendMessageForLater,     batchSendMessageWithHeaders,     batchSendMessageWithHeadersForLater,+    batchSendTopic,+    batchSendTopicForLater,+    batchSendTopicWithHeaders,+    batchSendTopicWithHeadersForLater,     batchSetVisibilityTimeoutAt,+    bindTopic,     changeVisibilityTimeout,     createPartitionedQueue,     createQueue,@@ -98,7 +147,10 @@     disableNotifyInsert,     dropQueue,     enableNotifyInsert,+    listNotifyInsertThrottles,     listQueues,+    listTopicBindings,+    listTopicBindingsForQueue,     pop,     queueMetrics,     readMessage,@@ -107,7 +159,14 @@     sendMessageForLater,     sendMessageWithHeaders,     sendMessageWithHeadersForLater,+    sendTopic,+    sendTopicWithHeaders,     setVisibilityTimeoutAt,+    testRouting,+    unbindTopic,+    updateNotifyInsert,+    validateRoutingKey,+    validateTopicPattern,   ) import Pgmq.Hasql.Statements.Types   ( BatchMessageQuery (..),@@ -115,8 +174,13 @@     BatchSendMessageForLater (..),     BatchSendMessageWithHeaders (..),     BatchSendMessageWithHeadersForLater (..),+    BatchSendTopic (..),+    BatchSendTopicForLater (..),+    BatchSendTopicWithHeaders (..),+    BatchSendTopicWithHeadersForLater (..),     BatchVisibilityTimeoutAtQuery (..),     BatchVisibilityTimeoutQuery (..),+    BindTopic (..),     CreatePartitionedQueue (..),     EnableNotifyInsert (..),     MessageQuery (..),@@ -128,6 +192,10 @@     SendMessageForLater (..),     SendMessageWithHeaders (..),     SendMessageWithHeadersForLater (..),+    SendTopic (..),+    SendTopicWithHeaders (..),+    UnbindTopic (..),+    UpdateNotifyInsert (..),     VisibilityTimeoutAtQuery (..),     VisibilityTimeoutQuery (..),   )@@ -136,8 +204,18 @@     MessageBody (..),     MessageHeaders (..),     MessageId (..),+    NotifyInsertThrottle (..),     Queue (..),     QueueName,+    RoutingKey,+    RoutingMatch (..),+    TopicBinding (..),+    TopicPattern,+    TopicSendResult (..),     parseQueueName,+    parseRoutingKey,+    parseTopicPattern,     queueNameToText,+    routingKeyToText,+    topicPatternToText,   )

@@ -3,6 +3,11 @@     messageIdDecoder,     queueDecoder,     queueMetricsDecoder,+    -- Topic decoders (pgmq 1.11.0+)+    topicBindingDecoder,+    routingMatchDecoder,+    topicSendResultDecoder,+    notifyInsertThrottleDecoder,   ) where @@ -10,7 +15,18 @@ import Data.Text (pack) import Hasql.Decoders qualified as D import Pgmq.Hasql.Statements.Types (QueueMetrics (..))-import Pgmq.Types (Message (..), MessageBody (..), MessageId (..), Queue (..), parseQueueName)+import Pgmq.Types+  ( Message (..),+    MessageBody (..),+    MessageId (..),+    NotifyInsertThrottle (..),+    Queue (..),+    RoutingMatch (..),+    TopicBinding (..),+    TopicSendResult (..),+    parseQueueName,+    parseTopicPattern,+  )  -- | Decoder for pgmq.message_record type -- Column order matches pgmq SQL: msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers@@ -58,3 +74,39 @@     <*> 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+)++-- | Decoder for topic binding records (pgmq 1.11.0+)+-- Column order: pattern, queue_name, bound_at, compiled_regex+topicBindingDecoder :: D.Row TopicBinding+topicBindingDecoder =+  TopicBinding+    <$> D.column (D.nonNullable $ D.refine (first (pack . show) . parseTopicPattern) D.text)+    <*> D.column (D.nonNullable D.text)+    <*> D.column (D.nonNullable D.timestamptz)+    <*> D.column (D.nonNullable D.text)++-- | Decoder for routing match results (pgmq 1.11.0+)+-- Column order: pattern, queue_name, compiled_regex+routingMatchDecoder :: D.Row RoutingMatch+routingMatchDecoder =+  RoutingMatch+    <$> D.column (D.nonNullable $ D.refine (first (pack . show) . parseTopicPattern) D.text)+    <*> D.column (D.nonNullable D.text)+    <*> D.column (D.nonNullable D.text)++-- | Decoder for topic send results (pgmq 1.11.0+)+-- Column order: queue_name, msg_id+topicSendResultDecoder :: D.Row TopicSendResult+topicSendResultDecoder =+  TopicSendResult+    <$> D.column (D.nonNullable D.text)+    <*> (MessageId <$> D.column (D.nonNullable D.int8))++-- | Decoder for notification throttle settings (pgmq 1.11.0+)+-- Column order: queue_name, throttle_interval_ms, last_notified_at+notifyInsertThrottleDecoder :: D.Row NotifyInsertThrottle+notifyInsertThrottleDecoder =+  NotifyInsertThrottle+    <$> D.column (D.nonNullable D.text)+    <*> D.column (D.nonNullable D.int4)+    <*> D.column (D.nonNullable D.timestamptz)

@@ -26,6 +26,18 @@     -- FIFO encoders (pgmq 1.8.0+)     readGroupedEncoder,     readGroupedWithPollEncoder,+    -- Topic encoders (pgmq 1.11.0+)+    routingKeyValue,+    topicPatternValue,+    bindTopicEncoder,+    unbindTopicEncoder,+    sendTopicEncoder,+    sendTopicWithHeadersEncoder,+    batchSendTopicEncoder,+    batchSendTopicForLaterEncoder,+    batchSendTopicWithHeadersEncoder,+    batchSendTopicWithHeadersForLaterEncoder,+    updateNotifyInsertEncoder,   ) where @@ -38,7 +50,11 @@     MessageHeaders (..),     MessageId (..),     QueueName,+    RoutingKey,+    TopicPattern,     queueNameToText,+    routingKeyToText,+    topicPatternToText,   )  queueNameEncoder :: E.Params QueueName@@ -214,3 +230,79 @@     <> (view #qty >$< E.param (E.nonNullable E.int4))     <> (view #maxPollSeconds >$< E.param (E.nonNullable E.int4))     <> (view #pollIntervalMs >$< E.param (E.nonNullable E.int4))++-- Topic encoders (pgmq 1.11.0+)++routingKeyValue :: E.Value RoutingKey+routingKeyValue = routingKeyToText >$< E.text++topicPatternValue :: E.Value TopicPattern+topicPatternValue = topicPatternToText >$< E.text++bindTopicEncoder :: E.Params BindTopic+bindTopicEncoder =+  (view #topicPattern >$< E.param (E.nonNullable topicPatternValue))+    <> (view #queueName >$< E.param (E.nonNullable queueNameValue))++unbindTopicEncoder :: E.Params UnbindTopic+unbindTopicEncoder =+  (view #topicPattern >$< E.param (E.nonNullable topicPatternValue))+    <> (view #queueName >$< E.param (E.nonNullable queueNameValue))++-- | Encoder for SendTopic (pgmq 1.11.0+)+-- SQL: pgmq.send_topic(routing_key, msg, coalesce(delay, 0))+sendTopicEncoder :: E.Params SendTopic+sendTopicEncoder =+  (view #routingKey >$< E.param (E.nonNullable routingKeyValue))+    <> (view #messageBody >$< E.param (E.nonNullable messageBodyValue))+    <> (view #delay >$< E.param (E.nullable E.int4))++-- | Encoder for SendTopicWithHeaders (pgmq 1.11.0+)+-- SQL: pgmq.send_topic(routing_key, msg, headers, coalesce(delay, 0))+sendTopicWithHeadersEncoder :: E.Params SendTopicWithHeaders+sendTopicWithHeadersEncoder =+  (view #routingKey >$< E.param (E.nonNullable routingKeyValue))+    <> (view #messageBody >$< E.param (E.nonNullable messageBodyValue))+    <> (view #messageHeaders >$< E.param (E.nonNullable messageHeadersValue))+    <> (view #delay >$< E.param (E.nullable E.int4))++-- | Encoder for BatchSendTopic (pgmq 1.11.0+)+-- SQL: pgmq.send_batch_topic(routing_key, msgs[], coalesce(delay, 0))+batchSendTopicEncoder :: E.Params BatchSendTopic+batchSendTopicEncoder =+  (view #routingKey >$< E.param (E.nonNullable routingKeyValue))+    <> (view #messageBodies >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageBodyValue))))))+    <> (view #delay >$< E.param (E.nullable E.int4))++-- | Encoder for BatchSendTopicForLater (pgmq 1.11.0+)+-- SQL: pgmq.send_batch_topic(routing_key, msgs[], timestamp)+batchSendTopicForLaterEncoder :: E.Params BatchSendTopicForLater+batchSendTopicForLaterEncoder =+  (view #routingKey >$< E.param (E.nonNullable routingKeyValue))+    <> (view #messageBodies >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageBodyValue))))))+    <> (view #scheduledAt >$< E.param (E.nonNullable E.timestamptz))++-- | Encoder for BatchSendTopicWithHeaders (pgmq 1.11.0+)+-- SQL: pgmq.send_batch_topic(routing_key, msgs[], headers[], coalesce(delay, 0))+batchSendTopicWithHeadersEncoder :: E.Params BatchSendTopicWithHeaders+batchSendTopicWithHeadersEncoder =+  (view #routingKey >$< E.param (E.nonNullable routingKeyValue))+    <> (view #messageBodies >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageBodyValue))))))+    <> (view #messageHeaders >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageHeadersValue))))))+    <> (view #delay >$< E.param (E.nullable E.int4))++-- | Encoder for BatchSendTopicWithHeadersForLater (pgmq 1.11.0+)+-- SQL: pgmq.send_batch_topic(routing_key, msgs[], headers[], timestamp)+batchSendTopicWithHeadersForLaterEncoder :: E.Params BatchSendTopicWithHeadersForLater+batchSendTopicWithHeadersForLaterEncoder =+  (view #routingKey >$< E.param (E.nonNullable routingKeyValue))+    <> (view #messageBodies >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageBodyValue))))))+    <> (view #messageHeaders >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageHeadersValue))))))+    <> (view #scheduledAt >$< E.param (E.nonNullable E.timestamptz))++-- | Encoder for UpdateNotifyInsert (pgmq 1.11.0+)+-- SQL: pgmq.update_notify_insert(queue_name, throttle_interval_ms)+updateNotifyInsertEncoder :: E.Params UpdateNotifyInsert+updateNotifyInsertEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #throttleIntervalMs >$< E.param (E.nonNullable E.int4))

@@ -39,6 +39,24 @@     -- FIFO index functions (pgmq 1.8.0+)     createFifoIndex,     createFifoIndexesAll,+    -- Topic management (pgmq 1.11.0+)+    bindTopic,+    unbindTopic,+    validateRoutingKey,+    validateTopicPattern,+    testRouting,+    listTopicBindings,+    listTopicBindingsForQueue,+    -- Topic sending (pgmq 1.11.0+)+    sendTopic,+    sendTopicWithHeaders,+    batchSendTopic,+    batchSendTopicForLater,+    batchSendTopicWithHeaders,+    batchSendTopicWithHeadersForLater,+    -- Notification management (pgmq 1.11.0+)+    listNotifyInsertThrottles,+    updateNotifyInsert,   ) where @@ -46,14 +64,20 @@ import Pgmq.Hasql.Prelude import Pgmq.Hasql.Statements qualified as Stmt import Pgmq.Hasql.Statements.Message qualified as Msg+import Pgmq.Hasql.Statements.TopicManagement qualified as Topic import Pgmq.Hasql.Statements.Types   ( BatchMessageQuery,     BatchSendMessage,     BatchSendMessageForLater,     BatchSendMessageWithHeaders,     BatchSendMessageWithHeadersForLater,+    BatchSendTopic,+    BatchSendTopicForLater,+    BatchSendTopicWithHeaders,+    BatchSendTopicWithHeadersForLater,     BatchVisibilityTimeoutAtQuery,     BatchVisibilityTimeoutQuery,+    BindTopic,     CreatePartitionedQueue,     EnableNotifyInsert,     MessageQuery,@@ -67,10 +91,25 @@     SendMessageForLater,     SendMessageWithHeaders,     SendMessageWithHeadersForLater,+    SendTopic,+    SendTopicWithHeaders,+    UnbindTopic,+    UpdateNotifyInsert,     VisibilityTimeoutAtQuery,     VisibilityTimeoutQuery,   )-import Pgmq.Types (Message, MessageId, Queue, QueueName)+import Pgmq.Types+  ( Message,+    MessageId,+    NotifyInsertThrottle,+    Queue,+    QueueName,+    RoutingKey,+    RoutingMatch,+    TopicBinding,+    TopicPattern,+    TopicSendResult,+  )  createQueue :: QueueName -> Session () createQueue q = statement q Stmt.createQueue@@ -196,3 +235,69 @@ -- | Create FIFO indexes for all queues (pgmq 1.8.0+) createFifoIndexesAll :: Session () createFifoIndexesAll = statement () Stmt.createFifoIndexesAll++-- Topic Management (pgmq 1.11.0+)++-- | Bind a topic pattern to a queue (pgmq 1.11.0+)+bindTopic :: BindTopic -> Session ()+bindTopic params = statement params Topic.bindTopic++-- | Unbind a topic pattern from a queue (pgmq 1.11.0+)+unbindTopic :: UnbindTopic -> Session Bool+unbindTopic params = statement params Topic.unbindTopic++-- | Validate a routing key (pgmq 1.11.0+)+validateRoutingKey :: RoutingKey -> Session Bool+validateRoutingKey key = statement key Topic.validateRoutingKey++-- | Validate a topic pattern (pgmq 1.11.0+)+validateTopicPattern :: TopicPattern -> Session Bool+validateTopicPattern pat = statement pat Topic.validateTopicPattern++-- | Test which queues a routing key would match (pgmq 1.11.0+)+testRouting :: RoutingKey -> Session [RoutingMatch]+testRouting key = statement key Topic.testRouting++-- | List all topic bindings (pgmq 1.11.0+)+listTopicBindings :: Session [TopicBinding]+listTopicBindings = statement () Topic.listTopicBindings++-- | List topic bindings for a specific queue (pgmq 1.11.0+)+listTopicBindingsForQueue :: QueueName -> Session [TopicBinding]+listTopicBindingsForQueue q = statement q Topic.listTopicBindingsForQueue++-- Topic Sending (pgmq 1.11.0+)++-- | Send a message via topic routing (pgmq 1.11.0+)+sendTopic :: SendTopic -> Session Int32+sendTopic msg = statement msg Msg.sendTopic++-- | Send a message via topic routing with headers (pgmq 1.11.0+)+sendTopicWithHeaders :: SendTopicWithHeaders -> Session Int32+sendTopicWithHeaders msg = statement msg Msg.sendTopicWithHeaders++-- | Batch send messages via topic routing (pgmq 1.11.0+)+batchSendTopic :: BatchSendTopic -> Session [TopicSendResult]+batchSendTopic msgs = statement msgs Msg.batchSendTopic++-- | Batch send messages via topic routing for later (pgmq 1.11.0+)+batchSendTopicForLater :: BatchSendTopicForLater -> Session [TopicSendResult]+batchSendTopicForLater msgs = statement msgs Msg.batchSendTopicForLater++-- | Batch send messages via topic routing with headers (pgmq 1.11.0+)+batchSendTopicWithHeaders :: BatchSendTopicWithHeaders -> Session [TopicSendResult]+batchSendTopicWithHeaders msgs = statement msgs Msg.batchSendTopicWithHeaders++-- | Batch send messages via topic routing with headers for later (pgmq 1.11.0+)+batchSendTopicWithHeadersForLater :: BatchSendTopicWithHeadersForLater -> Session [TopicSendResult]+batchSendTopicWithHeadersForLater msgs = statement msgs Msg.batchSendTopicWithHeadersForLater++-- Notification Management (pgmq 1.11.0+)++-- | List all notification insert throttle settings (pgmq 1.11.0+)+listNotifyInsertThrottles :: Session [NotifyInsertThrottle]+listNotifyInsertThrottles = statement () Stmt.listNotifyInsertThrottles++-- | Update the throttle interval for a queue's insert notifications (pgmq 1.11.0+)+updateNotifyInsert :: UpdateNotifyInsert -> Session ()+updateNotifyInsert params = statement params Stmt.updateNotifyInsert

@@ -2,9 +2,11 @@   ( module Pgmq.Hasql.Statements.Message,     module Pgmq.Hasql.Statements.QueueManagement,     module Pgmq.Hasql.Statements.QueueObservability,+    module Pgmq.Hasql.Statements.TopicManagement,   ) where  import Pgmq.Hasql.Statements.Message import Pgmq.Hasql.Statements.QueueManagement import Pgmq.Hasql.Statements.QueueObservability+import Pgmq.Hasql.Statements.TopicManagement

@@ -26,18 +26,29 @@     -- Round-robin FIFO functions (pgmq 1.9.0+)     readGroupedRoundRobin,     readGroupedRoundRobinWithPoll,+    -- Topic send functions (pgmq 1.11.0+)+    sendTopic,+    sendTopicWithHeaders,+    batchSendTopic,+    batchSendTopicForLater,+    batchSendTopicWithHeaders,+    batchSendTopicWithHeadersForLater,   ) where  import Hasql.Decoders qualified as D import Hasql.Statement (Statement, preparable)-import Pgmq.Hasql.Decoders (messageDecoder, messageIdDecoder)+import Pgmq.Hasql.Decoders (messageDecoder, messageIdDecoder, topicSendResultDecoder) import Pgmq.Hasql.Encoders   ( batchMessageQueryEncoder,     batchSendMessageEncoder,     batchSendMessageForLaterEncoder,     batchSendMessageWithHeadersEncoder,     batchSendMessageWithHeadersForLaterEncoder,+    batchSendTopicEncoder,+    batchSendTopicForLaterEncoder,+    batchSendTopicWithHeadersEncoder,+    batchSendTopicWithHeadersForLaterEncoder,     batchVisibilityTimeoutAtQueryEncoder,     batchVisibilityTimeoutQueryEncoder,     messageQueryEncoder,@@ -51,6 +62,8 @@     sendMessageForLaterEncoder,     sendMessageWithHeadersEncoder,     sendMessageWithHeadersForLaterEncoder,+    sendTopicEncoder,+    sendTopicWithHeadersEncoder,     visibilityTimeoutAtQueryEncoder,     visibilityTimeoutQueryEncoder,   )@@ -61,6 +74,10 @@     BatchSendMessageForLater,     BatchSendMessageWithHeaders,     BatchSendMessageWithHeadersForLater,+    BatchSendTopic,+    BatchSendTopicForLater,+    BatchSendTopicWithHeaders,+    BatchSendTopicWithHeadersForLater,     BatchVisibilityTimeoutAtQuery,     BatchVisibilityTimeoutQuery,     MessageQuery,@@ -73,10 +90,12 @@     SendMessageForLater,     SendMessageWithHeaders,     SendMessageWithHeadersForLater,+    SendTopic,+    SendTopicWithHeaders,     VisibilityTimeoutAtQuery,     VisibilityTimeoutQuery,   )-import Pgmq.Types (Message, MessageId, QueueName)+import Pgmq.Types (Message, MessageId, QueueName, TopicSendResult)  -- https://tembo.io/pgmq/api/sql/functions/#send -- Note: coalesce handles null delay to ensure correct function overload resolution@@ -270,3 +289,47 @@   where     sql = "select * from pgmq.read_grouped_rr_with_poll($1,$2,$3,$4,$5)"     decoder = D.rowVector messageDecoder++-- | Send a message via topic routing (pgmq 1.11.0+)+-- Returns the count of queues the message was delivered to.+sendTopic :: Statement SendTopic Int32+sendTopic = preparable sql sendTopicEncoder decoder+  where+    sql = "select pgmq.send_topic($1, $2, coalesce($3, 0))"+    decoder = D.singleRow (D.column (D.nonNullable D.int4))++-- | Send a message via topic routing with headers (pgmq 1.11.0+)+-- Returns the count of queues the message was delivered to.+sendTopicWithHeaders :: Statement SendTopicWithHeaders Int32+sendTopicWithHeaders = preparable sql sendTopicWithHeadersEncoder decoder+  where+    sql = "select pgmq.send_topic($1, $2, $3, coalesce($4, 0))"+    decoder = D.singleRow (D.column (D.nonNullable D.int4))++-- | Batch send messages via topic routing (pgmq 1.11.0+)+batchSendTopic :: Statement BatchSendTopic [TopicSendResult]+batchSendTopic = preparable sql batchSendTopicEncoder decoder+  where+    sql = "select * from pgmq.send_batch_topic($1, $2::jsonb[], coalesce($3, 0))"+    decoder = D.rowList topicSendResultDecoder++-- | Batch send messages via topic routing for later (pgmq 1.11.0+)+batchSendTopicForLater :: Statement BatchSendTopicForLater [TopicSendResult]+batchSendTopicForLater = preparable sql batchSendTopicForLaterEncoder decoder+  where+    sql = "select * from pgmq.send_batch_topic($1, $2::jsonb[], $3)"+    decoder = D.rowList topicSendResultDecoder++-- | Batch send messages via topic routing with headers (pgmq 1.11.0+)+batchSendTopicWithHeaders :: Statement BatchSendTopicWithHeaders [TopicSendResult]+batchSendTopicWithHeaders = preparable sql batchSendTopicWithHeadersEncoder decoder+  where+    sql = "select * from pgmq.send_batch_topic($1, $2::jsonb[], $3::jsonb[], coalesce($4, 0))"+    decoder = D.rowList topicSendResultDecoder++-- | Batch send messages via topic routing with headers for later (pgmq 1.11.0+)+batchSendTopicWithHeadersForLater :: Statement BatchSendTopicWithHeadersForLater [TopicSendResult]+batchSendTopicWithHeadersForLater = preparable sql batchSendTopicWithHeadersForLaterEncoder decoder+  where+    sql = "select * from pgmq.send_batch_topic($1, $2::jsonb[], $3::jsonb[], $4)"+    decoder = D.rowList topicSendResultDecoder

@@ -9,15 +9,19 @@     -- FIFO index functions (pgmq 1.8.0+)     createFifoIndex,     createFifoIndexesAll,+    -- Notification management (pgmq 1.11.0+)+    listNotifyInsertThrottles,+    updateNotifyInsert,   ) where  import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable)-import Pgmq.Hasql.Encoders (createPartitionedQueueEncoder, enableNotifyInsertEncoder, queueNameEncoder)-import Pgmq.Hasql.Statements.Types (CreatePartitionedQueue, EnableNotifyInsert)-import Pgmq.Types (QueueName)+import Pgmq.Hasql.Decoders (notifyInsertThrottleDecoder)+import Pgmq.Hasql.Encoders (createPartitionedQueueEncoder, enableNotifyInsertEncoder, queueNameEncoder, updateNotifyInsertEncoder)+import Pgmq.Hasql.Statements.Types (CreatePartitionedQueue, EnableNotifyInsert, UpdateNotifyInsert)+import Pgmq.Types (NotifyInsertThrottle, QueueName)  -- https://tembo.io/pgmq/api/sql/functions/#create createQueue :: Statement QueueName ()@@ -76,3 +80,16 @@ createFifoIndexesAll = preparable sql E.noParams D.noResult   where     sql = "select from pgmq.create_fifo_indexes_all()"++-- | List all notification insert throttle settings (pgmq 1.11.0+)+listNotifyInsertThrottles :: Statement () [NotifyInsertThrottle]+listNotifyInsertThrottles = preparable sql E.noParams decoder+  where+    sql = "select * from pgmq.list_notify_insert_throttles()"+    decoder = D.rowList notifyInsertThrottleDecoder++-- | Update the throttle interval for a queue's insert notifications (pgmq 1.11.0+)+updateNotifyInsert :: Statement UpdateNotifyInsert ()+updateNotifyInsert = preparable sql updateNotifyInsertEncoder D.noResult+  where+    sql = "select from pgmq.update_notify_insert($1, $2)"

@@ -0,0 +1,78 @@+-- | Topic management statements (pgmq 1.11.0+)+-- Provides AMQP-like topic-based routing: bind/unbind patterns to queues,+-- validate routing keys and patterns, test routing, and list bindings.+module Pgmq.Hasql.Statements.TopicManagement+  ( bindTopic,+    unbindTopic,+    validateRoutingKey,+    validateTopicPattern,+    testRouting,+    listTopicBindings,+    listTopicBindingsForQueue,+  )+where++import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (Statement, preparable)+import Pgmq.Hasql.Decoders (routingMatchDecoder, topicBindingDecoder)+import Pgmq.Hasql.Encoders+  ( bindTopicEncoder,+    queueNameEncoder,+    routingKeyValue,+    topicPatternValue,+    unbindTopicEncoder,+  )+import Pgmq.Hasql.Statements.Types (BindTopic, UnbindTopic)+import Pgmq.Types (QueueName, RoutingKey, RoutingMatch, TopicBinding, TopicPattern)++-- | Bind a topic pattern to a queue (pgmq 1.11.0+)+bindTopic :: Statement BindTopic ()+bindTopic = preparable sql bindTopicEncoder D.noResult+  where+    sql = "select from pgmq.bind_topic($1, $2)"++-- | Unbind a topic pattern from a queue (pgmq 1.11.0+)+unbindTopic :: Statement UnbindTopic Bool+unbindTopic = preparable sql unbindTopicEncoder decoder+  where+    sql = "select pgmq.unbind_topic($1, $2)"+    decoder = D.singleRow (D.column (D.nonNullable D.bool))++-- | Validate a routing key (pgmq 1.11.0+)+validateRoutingKey :: Statement RoutingKey Bool+validateRoutingKey = preparable sql encoder decoder+  where+    sql = "select pgmq.validate_routing_key($1)"+    encoder = E.param (E.nonNullable routingKeyValue)+    decoder = D.singleRow (D.column (D.nonNullable D.bool))++-- | Validate a topic pattern (pgmq 1.11.0+)+validateTopicPattern :: Statement TopicPattern Bool+validateTopicPattern = preparable sql encoder decoder+  where+    sql = "select pgmq.validate_topic_pattern($1)"+    encoder = E.param (E.nonNullable topicPatternValue)+    decoder = D.singleRow (D.column (D.nonNullable D.bool))++-- | Test which queues a routing key would match (pgmq 1.11.0+)+testRouting :: Statement RoutingKey [RoutingMatch]+testRouting = preparable sql encoder decoder+  where+    sql = "select * from pgmq.test_routing($1)"+    encoder = E.param (E.nonNullable routingKeyValue)+    decoder = D.rowList routingMatchDecoder++-- | List all topic bindings (pgmq 1.11.0+)+listTopicBindings :: Statement () [TopicBinding]+listTopicBindings = preparable sql E.noParams decoder+  where+    sql = "select * from pgmq.list_topic_bindings()"+    decoder = D.rowList topicBindingDecoder++-- | List topic bindings for a specific queue (pgmq 1.11.0+)+listTopicBindingsForQueue :: Statement QueueName [TopicBinding]+listTopicBindingsForQueue = preparable sql queueNameEncoder decoder+  where+    sql = "select * from pgmq.list_topic_bindings($1)"+    decoder = D.rowList topicBindingDecoder

@@ -23,12 +23,22 @@     -- FIFO read types (pgmq 1.8.0+)     ReadGrouped (..),     ReadGroupedWithPoll (..),+    -- Topic types (pgmq 1.11.0+)+    BindTopic (..),+    UnbindTopic (..),+    SendTopic (..),+    SendTopicWithHeaders (..),+    BatchSendTopic (..),+    BatchSendTopicForLater (..),+    BatchSendTopicWithHeaders (..),+    BatchSendTopicWithHeadersForLater (..),+    UpdateNotifyInsert (..),   ) where  import Data.Aeson (Value) import Pgmq.Hasql.Prelude-import Pgmq.Types (MessageBody, MessageHeaders, MessageId, QueueName)+import Pgmq.Types (MessageBody, MessageHeaders, MessageId, QueueName, RoutingKey, TopicPattern)  type Delay = Int32 @@ -216,5 +226,77 @@     qty :: !Int32,     maxPollSeconds :: !Int32,     pollIntervalMs :: !Int32+  }+  deriving stock (Generic)++-- | Bind a topic pattern to a queue (pgmq 1.11.0+)+data BindTopic = BindTopic+  { topicPattern :: !TopicPattern,+    queueName :: !QueueName+  }+  deriving stock (Generic)++-- | Unbind a topic pattern from a queue (pgmq 1.11.0+)+data UnbindTopic = UnbindTopic+  { topicPattern :: !TopicPattern,+    queueName :: !QueueName+  }+  deriving stock (Generic)++-- | Send a message via topic routing (pgmq 1.11.0+)+data SendTopic = SendTopic+  { routingKey :: !RoutingKey,+    messageBody :: !MessageBody,+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++-- | Send a message via topic routing with headers (pgmq 1.11.0+)+data SendTopicWithHeaders = SendTopicWithHeaders+  { routingKey :: !RoutingKey,+    messageBody :: !MessageBody,+    messageHeaders :: !MessageHeaders,+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++-- | Batch send messages via topic routing (pgmq 1.11.0+)+data BatchSendTopic = BatchSendTopic+  { routingKey :: !RoutingKey,+    messageBodies :: ![MessageBody],+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++-- | Batch send messages via topic routing for later (pgmq 1.11.0+)+data BatchSendTopicForLater = BatchSendTopicForLater+  { routingKey :: !RoutingKey,+    messageBodies :: ![MessageBody],+    scheduledAt :: !UTCTime+  }+  deriving stock (Generic)++-- | Batch send messages via topic routing with headers (pgmq 1.11.0+)+data BatchSendTopicWithHeaders = BatchSendTopicWithHeaders+  { routingKey :: !RoutingKey,+    messageBodies :: ![MessageBody],+    messageHeaders :: ![MessageHeaders],+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++-- | Batch send messages via topic routing with headers for later (pgmq 1.11.0+)+data BatchSendTopicWithHeadersForLater = BatchSendTopicWithHeadersForLater+  { routingKey :: !RoutingKey,+    messageBodies :: ![MessageBody],+    messageHeaders :: ![MessageHeaders],+    scheduledAt :: !UTCTime+  }+  deriving stock (Generic)++-- | Update notification throttle interval (pgmq 1.11.0+)+data UpdateNotifyInsert = UpdateNotifyInsert+  { queueName :: !QueueName,+    throttleIntervalMs :: !Int32   }   deriving stock (Generic)

@@ -11,6 +11,7 @@ import RoundTripSpec qualified import SchemaSpec qualified import Test.Tasty (defaultMain, testGroup)+import TopicSpec qualified  main :: IO () main = do@@ -25,7 +26,8 @@               SchemaSpec.tests pool,               RoundTripSpec.tests pool,               DecoderValidationSpec.tests pool,-              AllFunctionsDecoderSpec.tests pool+              AllFunctionsDecoderSpec.tests pool,+              TopicSpec.tests pool             ]     defaultMain tree   case result of

@@ -0,0 +1,347 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests for topic routing operations (pgmq 1.11.0+):+-- - bind/unbind topic+-- - send_topic / send_topic with headers+-- - batch send_topic (immediate, delayed, with headers)+-- - validate_routing_key / validate_topic_pattern+-- - test_routing+-- - list_topic_bindings+-- - list_notify_insert_throttles / update_notify_insert+module TopicSpec (tests) where++import Data.Aeson (object, (.=))+import Data.Text qualified as T+import Data.Time.Clock (addUTCTime, getCurrentTime)+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( BatchSendTopic (..),+    BatchSendTopicForLater (..),+    BatchSendTopicWithHeaders (..),+    BatchSendTopicWithHeadersForLater (..),+    BindTopic (..),+    EnableNotifyInsert (..),+    SendTopic (..),+    SendTopicWithHeaders (..),+    UnbindTopic (..),+    UpdateNotifyInsert (..),+  )+import Pgmq.Types+  ( MessageBody (..),+    MessageHeaders (..),+    QueueName,+    RoutingKey,+    TopicPattern,+    parseRoutingKey,+    parseTopicPattern,+  )+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, testCase)+import TestUtils (assertSession, cleanupQueue)++-- | All topic operation tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Topic Operations (pgmq 1.11.0+)"+    [ testBindUnbindTopic p,+      testSendTopic p,+      testSendTopicWithHeaders p,+      testBatchSendTopic p,+      testBatchSendTopicForLater p,+      testBatchSendTopicWithHeaders p,+      testBatchSendTopicWithHeadersForLater p,+      testTestRouting p,+      testValidateRoutingKey p,+      testValidateTopicPattern p,+      testListTopicBindings p,+      testListTopicBindingsForQueue p,+      testNotifyInsertThrottle p+    ]++mkRoutingKey :: T.Text -> RoutingKey+mkRoutingKey t = case parseRoutingKey t of+  Left err -> error $ "Invalid routing key: " <> show err+  Right rk -> rk++mkTopicPattern :: T.Text -> TopicPattern+mkTopicPattern t = case parseTopicPattern t of+  Left err -> error $ "Invalid topic pattern: " <> show err+  Right tp -> tp++-- | Helper to unbind a pattern and then cleanup the queue+unbindAndCleanup :: Pool.Pool -> TopicPattern -> QueueName -> IO ()+unbindAndCleanup pool pat qn = do+  _ <-+    assertSession pool $+      Sessions.unbindTopic+        UnbindTopic {topicPattern = pat, queueName = qn}+  cleanupQueue pool qn++-- | Test bind and unbind topic+testBindUnbindTopic :: Pool.Pool -> TestTree+testBindUnbindTopic p = testCase "bindTopic and unbindTopic" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Bind a pattern+    let pat = mkTopicPattern "bind-test.#"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    -- Verify binding exists for this queue+    bindings <- assertSession pool $ Sessions.listTopicBindingsForQueue queueName+    assertEqual "Should have 1 binding for queue" 1 (length bindings)+    -- Unbind+    removed <-+      assertSession pool $+        Sessions.unbindTopic+          UnbindTopic {topicPattern = pat, queueName = queueName}+    assertBool "Unbind should return True" removed+    -- Unbind again should return False+    removedAgain <-+      assertSession pool $+        Sessions.unbindTopic+          UnbindTopic {topicPattern = pat, queueName = queueName}+    assertBool "Second unbind should return False" (not removedAgain)+    cleanupQueue pool queueName++-- | Test sendTopic routes to matching queues+testSendTopic :: Pool.Pool -> TestTree+testSendTopic p = testCase "sendTopic routes message to matching queues" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Use a unique pattern prefix to avoid matching stale bindings+    let pat = mkTopicPattern "sendtest.#"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    -- Send via topic with routing key matching only our pattern+    matchedCount <-+      assertSession pool $+        Sessions.sendTopic+          SendTopic+            { routingKey = mkRoutingKey "sendtest.error",+              messageBody = MessageBody (object ["level" .= ("error" :: String)]),+              delay = Nothing+            }+    assertEqual "Should match 1 queue" 1 matchedCount+    unbindAndCleanup pool pat queueName++-- | Test sendTopicWithHeaders+testSendTopicWithHeaders :: Pool.Pool -> TestTree+testSendTopicWithHeaders p = testCase "sendTopicWithHeaders routes with headers" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let pat = mkTopicPattern "hdrstest.#"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    matchedCount <-+      assertSession pool $+        Sessions.sendTopicWithHeaders+          SendTopicWithHeaders+            { routingKey = mkRoutingKey "hdrstest.user.login",+              messageBody = MessageBody (object ["event" .= ("login" :: String)]),+              messageHeaders = MessageHeaders (object ["source" .= ("auth" :: String)]),+              delay = Nothing+            }+    assertEqual "Should match 1 queue" 1 matchedCount+    unbindAndCleanup pool pat queueName++-- | Test batchSendTopic+testBatchSendTopic :: Pool.Pool -> TestTree+testBatchSendTopic p = testCase "batchSendTopic routes batch to matching queues" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let pat = mkTopicPattern "batchtest.*"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    results <-+      assertSession pool $+        Sessions.batchSendTopic+          BatchSendTopic+            { routingKey = mkRoutingKey "batchtest.error",+              messageBodies =+                [ MessageBody (object ["msg" .= (1 :: Int)]),+                  MessageBody (object ["msg" .= (2 :: Int)])+                ],+              delay = Nothing+            }+    assertEqual "Should return 2 results (2 msgs to 1 queue)" 2 (length results)+    unbindAndCleanup pool pat queueName++-- | Test batchSendTopicForLater with a scheduled time+testBatchSendTopicForLater :: Pool.Pool -> TestTree+testBatchSendTopicForLater p = testCase "batchSendTopicForLater schedules batch via topic" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let pat = mkTopicPattern "batchlater.*"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    futureTime <- addUTCTime 60 <$> getCurrentTime+    results <-+      assertSession pool $+        Sessions.batchSendTopicForLater+          BatchSendTopicForLater+            { routingKey = mkRoutingKey "batchlater.info",+              messageBodies =+                [ MessageBody (object ["msg" .= (1 :: Int)]),+                  MessageBody (object ["msg" .= (2 :: Int)])+                ],+              scheduledAt = futureTime+            }+    assertEqual "Should return 2 results (2 msgs to 1 queue)" 2 (length results)+    unbindAndCleanup pool pat queueName++-- | Test batchSendTopicWithHeaders+testBatchSendTopicWithHeaders :: Pool.Pool -> TestTree+testBatchSendTopicWithHeaders p = testCase "batchSendTopicWithHeaders routes batch with headers" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let pat = mkTopicPattern "batchhdrs.*"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    results <-+      assertSession pool $+        Sessions.batchSendTopicWithHeaders+          BatchSendTopicWithHeaders+            { routingKey = mkRoutingKey "batchhdrs.warn",+              messageBodies =+                [ MessageBody (object ["msg" .= (1 :: Int)]),+                  MessageBody (object ["msg" .= (2 :: Int)])+                ],+              messageHeaders =+                [ MessageHeaders (object ["priority" .= ("high" :: String)]),+                  MessageHeaders (object ["priority" .= ("low" :: String)])+                ],+              delay = Nothing+            }+    assertEqual "Should return 2 results (2 msgs to 1 queue)" 2 (length results)+    unbindAndCleanup pool pat queueName++-- | Test batchSendTopicWithHeadersForLater+testBatchSendTopicWithHeadersForLater :: Pool.Pool -> TestTree+testBatchSendTopicWithHeadersForLater p = testCase "batchSendTopicWithHeadersForLater schedules batch with headers" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let pat = mkTopicPattern "batchhdrslater.*"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    futureTime <- addUTCTime 60 <$> getCurrentTime+    results <-+      assertSession pool $+        Sessions.batchSendTopicWithHeadersForLater+          BatchSendTopicWithHeadersForLater+            { routingKey = mkRoutingKey "batchhdrslater.error",+              messageBodies =+                [ MessageBody (object ["msg" .= (1 :: Int)]),+                  MessageBody (object ["msg" .= (2 :: Int)])+                ],+              messageHeaders =+                [ MessageHeaders (object ["source" .= ("app" :: String)]),+                  MessageHeaders (object ["source" .= ("sys" :: String)])+                ],+              scheduledAt = futureTime+            }+    assertEqual "Should return 2 results (2 msgs to 1 queue)" 2 (length results)+    unbindAndCleanup pool pat queueName++-- | Test validateRoutingKey+testValidateRoutingKey :: Pool.Pool -> TestTree+testValidateRoutingKey p = testCase "validateRoutingKey validates on server" $ do+  withTestFixture p $ \TestFixture {pool} -> do+    result <- assertSession pool $ Sessions.validateRoutingKey (mkRoutingKey "orders.created")+    assertBool "Valid routing key should return True" result++-- | Test validateTopicPattern+testValidateTopicPattern :: Pool.Pool -> TestTree+testValidateTopicPattern p = testCase "validateTopicPattern validates on server" $ do+  withTestFixture p $ \TestFixture {pool} -> do+    result <- assertSession pool $ Sessions.validateTopicPattern (mkTopicPattern "orders.#")+    assertBool "Valid topic pattern should return True" result++-- | Test testRouting dry-run+testTestRouting :: Pool.Pool -> TestTree+testTestRouting p = testCase "testRouting shows matching patterns" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Bind two patterns that both match "routetest.error"+    let pat1 = mkTopicPattern "routetest.#"+        pat2 = mkTopicPattern "routetest.*"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat1, queueName = queueName}+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat2, queueName = queueName}+    -- Test routing - both patterns should match+    matches <- assertSession pool $ Sessions.testRouting (mkRoutingKey "routetest.error")+    assertEqual "Should match 2 patterns" 2 (length matches)+    -- Cleanup+    _ <-+      assertSession pool $+        Sessions.unbindTopic+          UnbindTopic {topicPattern = pat1, queueName = queueName}+    unbindAndCleanup pool pat2 queueName++-- | Test listTopicBindings+testListTopicBindings :: Pool.Pool -> TestTree+testListTopicBindings p = testCase "listTopicBindings returns all bindings" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let pat = mkTopicPattern "listtest.#"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat, queueName = queueName}+    bindings <- assertSession pool Sessions.listTopicBindings+    assertBool "Should have at least 1 binding" (not $ null bindings)+    unbindAndCleanup pool pat queueName++-- | Test listTopicBindingsForQueue+testListTopicBindingsForQueue :: Pool.Pool -> TestTree+testListTopicBindingsForQueue p = testCase "listTopicBindingsForQueue returns queue-specific bindings" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let pat1 = mkTopicPattern "listqtest.events.#"+        pat2 = mkTopicPattern "listqtest.logs.*"+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat1, queueName = queueName}+    assertSession pool $+      Sessions.bindTopic+        BindTopic {topicPattern = pat2, queueName = queueName}+    bindings <- assertSession pool $ Sessions.listTopicBindingsForQueue queueName+    assertEqual "Should have 2 bindings for queue" 2 (length bindings)+    -- Cleanup+    _ <-+      assertSession pool $+        Sessions.unbindTopic+          UnbindTopic {topicPattern = pat1, queueName = queueName}+    unbindAndCleanup pool pat2 queueName++-- | Test listNotifyInsertThrottles and updateNotifyInsert+testNotifyInsertThrottle :: Pool.Pool -> TestTree+testNotifyInsertThrottle p = testCase "listNotifyInsertThrottles and updateNotifyInsert" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Enable notifications first+    assertSession pool $+      Sessions.enableNotifyInsert+        EnableNotifyInsert {queueName = queueName, throttleIntervalMs = Just 100}+    -- List throttles+    throttles <- assertSession pool Sessions.listNotifyInsertThrottles+    assertBool "Should have at least 1 throttle entry" (not $ null throttles)+    -- Update throttle+    assertSession pool $+      Sessions.updateNotifyInsert+        UpdateNotifyInsert {queueName = queueName, throttleIntervalMs = 500}+    -- Verify update+    throttlesAfter <- assertSession pool Sessions.listNotifyInsertThrottles+    assertBool "Should still have throttle entries" (not $ null throttlesAfter)+    cleanupQueue pool queueName