diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/pgmq-hasql.cabal b/pgmq-hasql.cabal --- a/pgmq-hasql.cabal +++ b/pgmq-hasql.cabal @@ -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 diff --git a/src/Pgmq.hs b/src/Pgmq.hs --- a/src/Pgmq.hs +++ b/src/Pgmq.hs @@ -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, ) diff --git a/src/Pgmq/Hasql/Decoders.hs b/src/Pgmq/Hasql/Decoders.hs --- a/src/Pgmq/Hasql/Decoders.hs +++ b/src/Pgmq/Hasql/Decoders.hs @@ -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) diff --git a/src/Pgmq/Hasql/Encoders.hs b/src/Pgmq/Hasql/Encoders.hs --- a/src/Pgmq/Hasql/Encoders.hs +++ b/src/Pgmq/Hasql/Encoders.hs @@ -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)) diff --git a/src/Pgmq/Hasql/Sessions.hs b/src/Pgmq/Hasql/Sessions.hs --- a/src/Pgmq/Hasql/Sessions.hs +++ b/src/Pgmq/Hasql/Sessions.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements.hs b/src/Pgmq/Hasql/Statements.hs --- a/src/Pgmq/Hasql/Statements.hs +++ b/src/Pgmq/Hasql/Statements.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements/Message.hs b/src/Pgmq/Hasql/Statements/Message.hs --- a/src/Pgmq/Hasql/Statements/Message.hs +++ b/src/Pgmq/Hasql/Statements/Message.hs @@ -26,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 diff --git a/src/Pgmq/Hasql/Statements/QueueManagement.hs b/src/Pgmq/Hasql/Statements/QueueManagement.hs --- a/src/Pgmq/Hasql/Statements/QueueManagement.hs +++ b/src/Pgmq/Hasql/Statements/QueueManagement.hs @@ -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)" diff --git a/src/Pgmq/Hasql/Statements/TopicManagement.hs b/src/Pgmq/Hasql/Statements/TopicManagement.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Statements/TopicManagement.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements/Types.hs b/src/Pgmq/Hasql/Statements/Types.hs --- a/src/Pgmq/Hasql/Statements/Types.hs +++ b/src/Pgmq/Hasql/Statements/Types.hs @@ -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) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 diff --git a/test/TopicSpec.hs b/test/TopicSpec.hs new file mode 100644 --- /dev/null +++ b/test/TopicSpec.hs @@ -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