pgmq-effectful 0.1.0.0 → 0.1.1.0

raw patch · 7 files changed

+330/−7 lines, 7 filesdep ~pgmq-coredep ~pgmq-hasqlPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: pgmq-core, pgmq-hasql

API changes (from Hackage documentation)

Files

• CHANGELOG.md +9/−0
• pgmq-effectful.cabal +4/−4
• src/Pgmq/Effectful.hs +78/−0
• src/Pgmq/Effectful/Effect.hs +129/−2
• src/Pgmq/Effectful/Interpreter.hs +18/−0
• src/Pgmq/Effectful/Interpreter/Traced.hs +87/−1
• src/Pgmq/Effectful/Telemetry.hs +5/−0

@@ -1,5 +1,14 @@ # Revision history for pgmq-effectful +## 0.1.1.0 -- 2026-02-23++### New Features++* Effectful effects and interpreters for pgmq 1.11.0 topic routing operations+* Topic management: `bindTopic`, `unbindTopic`, `validateRoutingKey`, `validateTopicPattern`, `testRouting`, `listTopicBindings`, `listTopicBindingsForQueue`+* Topic sending: `sendTopic`, `sendTopicWithHeaders`, `batchSendTopic`, `batchSendTopicForLater`, `batchSendTopicWithHeaders`, `batchSendTopicWithHeadersForLater`+* Notification management: `listNotifyInsertThrottles`, `updateNotifyInsert`+ ## 0.1.0.0 -- 2026-02-21  * Initial release

@@ -1,6 +1,6 @@ cabal-version:   3.4 name:            pgmq-effectful-version:         0.1.0.0+version:         0.1.1.0 synopsis:        Effectful effects for PGMQ (PostgreSQL Message Queue) description:   Effectful effects and interpreters for pgmq-hs, a Haskell client@@ -11,7 +11,7 @@ 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@@ -55,8 +55,8 @@     , hasql-pool                       ^>=1.4     , hs-opentelemetry-api             >=0.2   && <0.4     , hs-opentelemetry-propagator-w3c  >=0.0   && <0.2-    , pgmq-core                        >=0.1   && <0.2-    , pgmq-hasql                       >=0.1   && <0.2+    , pgmq-core                        >=0.1.1 && <0.2+    , pgmq-hasql                       >=0.1.1 && <0.2     , text                             >=2.0   && <2.2     , unliftio                         >=0.2   && <0.3     , unordered-containers             >=0.2   && <0.3

@@ -56,6 +56,29 @@     readWithPoll,     pop, +    -- * 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,+     -- * Queue Observability     listQueues,     queueMetrics,@@ -89,6 +112,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,@@ -106,6 +150,11 @@     batchSendMessageForLater,     batchSendMessageWithHeaders,     batchSendMessageWithHeadersForLater,+    batchSendTopic,+    batchSendTopicForLater,+    batchSendTopicWithHeaders,+    batchSendTopicWithHeadersForLater,+    bindTopic,     changeVisibilityTimeout,     createPartitionedQueue,     createQueue,@@ -116,7 +165,10 @@     disableNotifyInsert,     dropQueue,     enableNotifyInsert,+    listNotifyInsertThrottles,     listQueues,+    listTopicBindings,+    listTopicBindingsForQueue,     pop,     queueMetrics,     readMessage,@@ -125,6 +177,13 @@     sendMessageForLater,     sendMessageWithHeaders,     sendMessageWithHeadersForLater,+    sendTopic,+    sendTopicWithHeaders,+    testRouting,+    unbindTopic,+    updateNotifyInsert,+    validateRoutingKey,+    validateTopicPattern,   ) import Pgmq.Effectful.Interpreter (PgmqError (..), runPgmq) import Pgmq.Effectful.Interpreter.Traced@@ -150,7 +209,12 @@     BatchSendMessageForLater (..),     BatchSendMessageWithHeaders (..),     BatchSendMessageWithHeadersForLater (..),+    BatchSendTopic (..),+    BatchSendTopicForLater (..),+    BatchSendTopicWithHeaders (..),+    BatchSendTopicWithHeadersForLater (..),     BatchVisibilityTimeoutQuery (..),+    BindTopic (..),     CreatePartitionedQueue (..),     EnableNotifyInsert (..),     MessageQuery (..),@@ -162,6 +226,10 @@     SendMessageForLater (..),     SendMessageWithHeaders (..),     SendMessageWithHeadersForLater (..),+    SendTopic (..),+    SendTopicWithHeaders (..),+    UnbindTopic (..),+    UpdateNotifyInsert (..),     VisibilityTimeoutQuery (..),   ) import Pgmq.Types@@ -169,8 +237,18 @@     MessageBody (..),     MessageHeaders (..),     MessageId (..),+    NotifyInsertThrottle (..),     Queue (..),     QueueName,+    RoutingKey,+    RoutingMatch (..),+    TopicBinding (..),+    TopicPattern,+    TopicSendResult (..),     parseQueueName,+    parseRoutingKey,+    parseTopicPattern,     queueNameToText,+    routingKeyToText,+    topicPatternToText,   )

@@ -51,6 +51,29 @@     readGroupedRoundRobin,     readGroupedRoundRobinWithPoll, +    -- * 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,+     -- * Queue Observability     listQueues,     queueMetrics,@@ -58,7 +81,7 @@   ) where -import Data.Int (Int64)+import Data.Int (Int32, Int64) import Data.Vector (Vector) import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>)) import Effectful.Dispatch.Dynamic (send)@@ -68,8 +91,13 @@     BatchSendMessageForLater,     BatchSendMessageWithHeaders,     BatchSendMessageWithHeadersForLater,+    BatchSendTopic,+    BatchSendTopicForLater,+    BatchSendTopicWithHeaders,+    BatchSendTopicWithHeadersForLater,     BatchVisibilityTimeoutAtQuery,     BatchVisibilityTimeoutQuery,+    BindTopic,     CreatePartitionedQueue,     EnableNotifyInsert,     MessageQuery,@@ -83,10 +111,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,+  )  -- | Effect for pgmq message queue operations. data Pgmq :: Effect where@@ -128,6 +171,24 @@   -- Round-robin FIFO Read (pgmq 1.9.0+)   ReadGroupedRoundRobin :: ReadGrouped -> Pgmq m (Vector Message)   ReadGroupedRoundRobinWithPoll :: ReadGroupedWithPoll -> Pgmq m (Vector Message)+  -- Topic Management (pgmq 1.11.0+)+  BindTopic :: BindTopic -> Pgmq m ()+  UnbindTopic :: UnbindTopic -> Pgmq m Bool+  ValidateRoutingKey :: RoutingKey -> Pgmq m Bool+  ValidateTopicPattern :: TopicPattern -> Pgmq m Bool+  TestRouting :: RoutingKey -> Pgmq m [RoutingMatch]+  ListTopicBindings :: Pgmq m [TopicBinding]+  ListTopicBindingsForQueue :: QueueName -> Pgmq m [TopicBinding]+  -- Topic Sending (pgmq 1.11.0+)+  SendTopic :: SendTopic -> Pgmq m Int32+  SendTopicWithHeaders :: SendTopicWithHeaders -> Pgmq m Int32+  BatchSendTopic :: BatchSendTopic -> Pgmq m [TopicSendResult]+  BatchSendTopicForLater :: BatchSendTopicForLater -> Pgmq m [TopicSendResult]+  BatchSendTopicWithHeaders :: BatchSendTopicWithHeaders -> Pgmq m [TopicSendResult]+  BatchSendTopicWithHeadersForLater :: BatchSendTopicWithHeadersForLater -> Pgmq m [TopicSendResult]+  -- Notification Management (pgmq 1.11.0+)+  ListNotifyInsertThrottles :: Pgmq m [NotifyInsertThrottle]+  UpdateNotifyInsert :: UpdateNotifyInsert -> Pgmq m ()   -- Queue Observability   ListQueues :: Pgmq m [Queue]   QueueMetrics :: QueueName -> Pgmq m QueueMetrics@@ -248,6 +309,72 @@ -- | Round-robin FIFO read with polling (pgmq 1.9.0+) readGroupedRoundRobinWithPoll :: (Pgmq :> es) => ReadGroupedWithPoll -> Eff es (Vector Message) readGroupedRoundRobinWithPoll = send . ReadGroupedRoundRobinWithPoll++-- Topic Management (pgmq 1.11.0+)++-- | Bind a topic pattern to a queue (pgmq 1.11.0+)+bindTopic :: (Pgmq :> es) => BindTopic -> Eff es ()+bindTopic = send . BindTopic++-- | Unbind a topic pattern from a queue (pgmq 1.11.0+)+unbindTopic :: (Pgmq :> es) => UnbindTopic -> Eff es Bool+unbindTopic = send . UnbindTopic++-- | Validate a routing key (pgmq 1.11.0+)+validateRoutingKey :: (Pgmq :> es) => RoutingKey -> Eff es Bool+validateRoutingKey = send . ValidateRoutingKey++-- | Validate a topic pattern (pgmq 1.11.0+)+validateTopicPattern :: (Pgmq :> es) => TopicPattern -> Eff es Bool+validateTopicPattern = send . ValidateTopicPattern++-- | Test which queues a routing key would match (pgmq 1.11.0+)+testRouting :: (Pgmq :> es) => RoutingKey -> Eff es [RoutingMatch]+testRouting = send . TestRouting++-- | List all topic bindings (pgmq 1.11.0+)+listTopicBindings :: (Pgmq :> es) => Eff es [TopicBinding]+listTopicBindings = send ListTopicBindings++-- | List topic bindings for a specific queue (pgmq 1.11.0+)+listTopicBindingsForQueue :: (Pgmq :> es) => QueueName -> Eff es [TopicBinding]+listTopicBindingsForQueue = send . ListTopicBindingsForQueue++-- Topic Sending (pgmq 1.11.0+)++-- | Send a message via topic routing (pgmq 1.11.0+)+sendTopic :: (Pgmq :> es) => SendTopic -> Eff es Int32+sendTopic = send . SendTopic++-- | Send a message via topic routing with headers (pgmq 1.11.0+)+sendTopicWithHeaders :: (Pgmq :> es) => SendTopicWithHeaders -> Eff es Int32+sendTopicWithHeaders = send . SendTopicWithHeaders++-- | Batch send messages via topic routing (pgmq 1.11.0+)+batchSendTopic :: (Pgmq :> es) => BatchSendTopic -> Eff es [TopicSendResult]+batchSendTopic = send . BatchSendTopic++-- | Batch send messages via topic routing for later (pgmq 1.11.0+)+batchSendTopicForLater :: (Pgmq :> es) => BatchSendTopicForLater -> Eff es [TopicSendResult]+batchSendTopicForLater = send . BatchSendTopicForLater++-- | Batch send messages via topic routing with headers (pgmq 1.11.0+)+batchSendTopicWithHeaders :: (Pgmq :> es) => BatchSendTopicWithHeaders -> Eff es [TopicSendResult]+batchSendTopicWithHeaders = send . BatchSendTopicWithHeaders++-- | Batch send messages via topic routing with headers for later (pgmq 1.11.0+)+batchSendTopicWithHeadersForLater :: (Pgmq :> es) => BatchSendTopicWithHeadersForLater -> Eff es [TopicSendResult]+batchSendTopicWithHeadersForLater = send . BatchSendTopicWithHeadersForLater++-- Notification Management (pgmq 1.11.0+)++-- | List all notification insert throttle settings (pgmq 1.11.0+)+listNotifyInsertThrottles :: (Pgmq :> es) => Eff es [NotifyInsertThrottle]+listNotifyInsertThrottles = send ListNotifyInsertThrottles++-- | Update the throttle interval for a queue's insert notifications (pgmq 1.11.0+)+updateNotifyInsert :: (Pgmq :> es) => UpdateNotifyInsert -> Eff es ()+updateNotifyInsert = send . UpdateNotifyInsert  -- Queue Observability 

@@ -67,6 +67,24 @@   -- Round-robin FIFO Read (pgmq 1.9.0+)   ReadGroupedRoundRobin query -> runSession pool $ Sessions.readGroupedRoundRobin query   ReadGroupedRoundRobinWithPoll query -> runSession pool $ Sessions.readGroupedRoundRobinWithPoll query+  -- Topic Management (pgmq 1.11.0+)+  BindTopic params -> runSession pool $ Sessions.bindTopic params+  UnbindTopic params -> runSession pool $ Sessions.unbindTopic params+  ValidateRoutingKey key -> runSession pool $ Sessions.validateRoutingKey key+  ValidateTopicPattern pat -> runSession pool $ Sessions.validateTopicPattern pat+  TestRouting key -> runSession pool $ Sessions.testRouting key+  ListTopicBindings -> runSession pool Sessions.listTopicBindings+  ListTopicBindingsForQueue q -> runSession pool $ Sessions.listTopicBindingsForQueue q+  -- Topic Sending (pgmq 1.11.0+)+  SendTopic msg -> runSession pool $ Sessions.sendTopic msg+  SendTopicWithHeaders msg -> runSession pool $ Sessions.sendTopicWithHeaders msg+  BatchSendTopic msgs -> runSession pool $ Sessions.batchSendTopic msgs+  BatchSendTopicForLater msgs -> runSession pool $ Sessions.batchSendTopicForLater msgs+  BatchSendTopicWithHeaders msgs -> runSession pool $ Sessions.batchSendTopicWithHeaders msgs+  BatchSendTopicWithHeadersForLater msgs -> runSession pool $ Sessions.batchSendTopicWithHeadersForLater msgs+  -- Notification Management (pgmq 1.11.0+)+  ListNotifyInsertThrottles -> runSession pool Sessions.listNotifyInsertThrottles+  UpdateNotifyInsert params -> runSession pool $ Sessions.updateNotifyInsert params   -- Queue Observability   ListQueues -> runSession pool Sessions.listQueues   QueueMetrics q -> runSession pool $ Sessions.queueMetrics q

@@ -55,7 +55,7 @@ import Pgmq.Effectful.Telemetry import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types qualified as Types-import Pgmq.Types (MessageId (..), QueueName, queueNameToText)+import Pgmq.Types (MessageId (..), QueueName, RoutingKey, queueNameToText, routingKeyToText)  -- | Configuration for OpenTelemetry tracing. data TracingConfig = TracingConfig@@ -198,6 +198,54 @@   BatchSetVisibilityTimeoutAt query@(Types.BatchVisibilityTimeoutAtQuery qn msgIds _) ->     withTracedSessionWithCount config "pgmq set_vt_at_batch" OTel.Internal qn (length msgIds) pool $       Sessions.batchSetVisibilityTimeoutAt query+  -- Topic Management (Internal spans, pgmq 1.11.0+)+  BindTopic params@(Types.BindTopic _ qn) ->+    withTracedSession config "pgmq bind_topic" OTel.Internal qn pool $+      Sessions.bindTopic params+  UnbindTopic params@(Types.UnbindTopic _ qn) ->+    withTracedSession config "pgmq unbind_topic" OTel.Internal qn pool $+      Sessions.unbindTopic params+  ValidateRoutingKey key ->+    withTracedSessionWithRoutingKey config "pgmq validate_routing_key" OTel.Internal key pool $+      Sessions.validateRoutingKey key+  ValidateTopicPattern _pat ->+    withTracedSessionNoQueue config "pgmq validate_topic_pattern" OTel.Internal pool $+      Sessions.validateTopicPattern _pat+  TestRouting key ->+    withTracedSessionWithRoutingKey config "pgmq test_routing" OTel.Internal key pool $+      Sessions.testRouting key+  ListTopicBindings ->+    withTracedSessionNoQueue config "pgmq list_topic_bindings" OTel.Internal pool $+      Sessions.listTopicBindings+  ListTopicBindingsForQueue q ->+    withTracedSession config "pgmq list_topic_bindings" OTel.Internal q pool $+      Sessions.listTopicBindingsForQueue q+  -- Topic Sending (Producer spans, pgmq 1.11.0+)+  SendTopic msg@(Types.SendTopic rk _ _) ->+    withTracedSessionWithRoutingKey config "pgmq send_topic" OTel.Producer rk pool $+      Sessions.sendTopic msg+  SendTopicWithHeaders msg@(Types.SendTopicWithHeaders rk _ _ _) ->+    withTracedSessionWithRoutingKey config "pgmq send_topic" OTel.Producer rk pool $+      Sessions.sendTopicWithHeaders msg+  BatchSendTopic msg@(Types.BatchSendTopic rk bodies _) ->+    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+      Sessions.batchSendTopic msg+  BatchSendTopicForLater msg@(Types.BatchSendTopicForLater rk bodies _) ->+    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+      Sessions.batchSendTopicForLater msg+  BatchSendTopicWithHeaders msg@(Types.BatchSendTopicWithHeaders rk bodies _ _) ->+    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+      Sessions.batchSendTopicWithHeaders msg+  BatchSendTopicWithHeadersForLater msg@(Types.BatchSendTopicWithHeadersForLater rk bodies _ _) ->+    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+      Sessions.batchSendTopicWithHeadersForLater msg+  -- Notification Management (Internal spans, pgmq 1.11.0+)+  ListNotifyInsertThrottles ->+    withTracedSessionNoQueue config "pgmq list_notify_insert_throttles" OTel.Internal pool $+      Sessions.listNotifyInsertThrottles+  UpdateNotifyInsert params@(Types.UpdateNotifyInsert qn _) ->+    withTracedSession config "pgmq update_notify_insert" OTel.Internal qn pool $+      Sessions.updateNotifyInsert params   -- Queue Observability (Internal spans)   ListQueues ->     withTracedSessionNoQueue config "pgmq list_queues" OTel.Internal pool $@@ -321,3 +369,41 @@ addBatchCountAttribute :: OTel.Span -> Int -> IO () addBatchCountAttribute s count =   OTel.addAttribute s messagingBatchMessageCount count++addRoutingKeyAttribute :: OTel.Span -> RoutingKey -> IO ()+addRoutingKeyAttribute s rk =+  OTel.addAttribute s messagingRoutingKey (routingKeyToText rk)++withTracedSessionWithRoutingKey ::+  (IOE :> es) =>+  TracingConfig ->+  Text ->+  OTel.SpanKind ->+  RoutingKey ->+  Pool ->+  Hasql.Session.Session a ->+  Eff es a+withTracedSessionWithRoutingKey config spanName kind routingKey pool session = do+  let args = OTel.defaultSpanArguments {OTel.kind = kind}+  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do+    addBaseAttributes s+    addRoutingKeyAttribute s routingKey+    runSessionIO config s pool session++withTracedSessionWithRoutingKeyAndCount ::+  (IOE :> es) =>+  TracingConfig ->+  Text ->+  OTel.SpanKind ->+  RoutingKey ->+  Int ->+  Pool ->+  Hasql.Session.Session a ->+  Eff es a+withTracedSessionWithRoutingKeyAndCount config spanName kind routingKey count pool session = do+  let args = OTel.defaultSpanArguments {OTel.kind = kind}+  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do+    addBaseAttributes s+    addRoutingKeyAttribute s routingKey+    addBatchCountAttribute s count+    runSessionIO config s pool session

@@ -17,6 +17,7 @@     messagingDestinationName,     messagingMessageId,     messagingBatchMessageCount,+    messagingRoutingKey,     dbSystem,     dbOperationName,   )@@ -99,6 +100,10 @@ -- | Number of messages in a batch operation. messagingBatchMessageCount :: Text messagingBatchMessageCount = "messaging.batch.message_count"++-- | The routing key for topic-based message routing.+messagingRoutingKey :: Text+messagingRoutingKey = "messaging.destination.routing_key"  -- | The database system. -- Value: "postgresql"