packages feed

pgmq-effectful (empty) → 0.1.0.0

raw patch · 9 files changed

+1175/−0 lines, 9 filesdep +aesondep +basedep +bytestring

Dependencies added: aeson, base, bytestring, effectful-core, hasql, hasql-pool, hs-opentelemetry-api, hs-opentelemetry-propagator-w3c, pgmq-core, pgmq-hasql, text, unliftio, unordered-containers, vector

Files

+ CHANGELOG.md view
@@ -0,0 +1,8 @@+# Revision history for pgmq-effectful++## 0.1.0.0 -- 2026-02-21++* Initial release+* Effectful effects and interpreters for all pgmq operations+* OpenTelemetry instrumentation via traced interpreter+* Support for pgmq 1.5.0 through 1.10.0 features
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2025 Nadeem Bitar++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ pgmq-effectful.cabal view
@@ -0,0 +1,66 @@+cabal-version:   3.4+name:            pgmq-effectful+version:         0.1.0.0+synopsis:        Effectful effects for PGMQ (PostgreSQL Message Queue)+description:+  Effectful effects and interpreters for pgmq-hs, a Haskell client+  library for PGMQ (PostgreSQL Message Queue). Includes OpenTelemetry+  instrumentation for tracing pgmq operations.++homepage:        https://github.com/topagentnetwork/pgmq-hs+license:         MIT+license-file:    LICENSE+author:          Nadeem Bitar+maintainer:      nadeem@topagentnetwork.com+category:        Database+build-type:      Simple+extra-doc-files: CHANGELOG.md++common warnings+  ghc-options:+    -Wall -Wcompat -Widentities -Wincomplete-uni-patterns+    -Wincomplete-record-updates -Wredundant-constraints+    -fhide-source-paths -Wmissing-export-lists -Wpartial-fields+    -Wmissing-deriving-strategies++library+  import:             warnings+  exposed-modules:+    Pgmq.Effectful+    Pgmq.Effectful.Effect+    Pgmq.Effectful.Interpreter+    Pgmq.Effectful.Interpreter.Traced+    Pgmq.Effectful.Telemetry+    Pgmq.Effectful.Traced++  default-extensions:+    DataKinds+    DuplicateRecordFields+    GADTs+    GeneralisedNewtypeDeriving+    ImportQualifiedPost+    LambdaCase+    NoFieldSelectors+    OverloadedRecordDot+    OverloadedStrings+    TypeFamilies+    TypeOperators++  build-depends:+    , aeson                            >=2.0   && <2.3+    , base                             >=4.18  && <5+    , bytestring                       >=0.11  && <0.13+    , effectful-core                   ^>=2.5  || ^>=2.6+    , hasql                            ^>=1.10+    , 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+    , text                             >=2.0   && <2.2+    , unliftio                         >=0.2   && <0.3+    , unordered-containers             >=0.2   && <0.3+    , vector                           ^>=0.13++  hs-source-dirs:     src+  default-language:   GHC2024
+ src/Pgmq/Effectful.hs view
@@ -0,0 +1,176 @@+module Pgmq.Effectful+  ( -- * Effect+    Pgmq,++    -- * Interpreters+    runPgmq,+    PgmqError (..),++    -- ** Traced Interpreters+    runPgmqTraced,+    runPgmqTracedWith,+    TracingConfig (..),+    defaultTracingConfig,++    -- ** Traced Operations+    sendMessageTraced,+    readMessageWithContext,+    MessageWithContext,++    -- ** Telemetry Utilities+    injectTraceContext,+    extractTraceContext,+    mergeTraceHeaders,+    TraceHeaders,++    -- * Queue Management+    createQueue,+    dropQueue,+    createPartitionedQueue,+    createUnloggedQueue,+    detachArchive,++    -- ** Notifications (pgmq 1.7.0+)+    enableNotifyInsert,+    disableNotifyInsert,++    -- * Message Operations+    sendMessage,+    sendMessageForLater,+    batchSendMessage,+    batchSendMessageForLater,++    -- ** With Headers (pgmq 1.5.0+)+    sendMessageWithHeaders,+    sendMessageWithHeadersForLater,+    batchSendMessageWithHeaders,+    batchSendMessageWithHeadersForLater,+    readMessage,+    deleteMessage,+    batchDeleteMessages,+    archiveMessage,+    batchArchiveMessages,+    deleteAllMessagesFromQueue,+    changeVisibilityTimeout,+    batchChangeVisibilityTimeout,+    readWithPoll,+    pop,++    -- * Queue Observability+    listQueues,+    queueMetrics,+    allQueueMetrics,++    -- * Types+    MessageBody (..),+    MessageHeaders (..),+    MessageId (..),+    Message (..),+    Queue (..),+    QueueName,+    SendMessage (..),+    SendMessageForLater (..),+    BatchSendMessage (..),+    BatchSendMessageForLater (..),++    -- ** With Headers (pgmq 1.5.0+)+    SendMessageWithHeaders (..),+    SendMessageWithHeadersForLater (..),+    BatchSendMessageWithHeaders (..),+    BatchSendMessageWithHeadersForLater (..),+    ReadMessage (..),+    PopMessage (..),+    EnableNotifyInsert (..),+    MessageQuery (..),+    BatchMessageQuery (..),+    VisibilityTimeoutQuery (..),+    BatchVisibilityTimeoutQuery (..),+    ReadWithPollMessage (..),+    CreatePartitionedQueue (..),+    QueueMetrics (..),++    -- * Queue Name Utilities+    parseQueueName,+    queueNameToText,+  )+where++import Pgmq.Effectful.Effect+  ( Pgmq,+    allQueueMetrics,+    archiveMessage,+    batchArchiveMessages,+    batchChangeVisibilityTimeout,+    batchDeleteMessages,+    batchSendMessage,+    batchSendMessageForLater,+    batchSendMessageWithHeaders,+    batchSendMessageWithHeadersForLater,+    changeVisibilityTimeout,+    createPartitionedQueue,+    createQueue,+    createUnloggedQueue,+    deleteAllMessagesFromQueue,+    deleteMessage,+    detachArchive,+    disableNotifyInsert,+    dropQueue,+    enableNotifyInsert,+    listQueues,+    pop,+    queueMetrics,+    readMessage,+    readWithPoll,+    sendMessage,+    sendMessageForLater,+    sendMessageWithHeaders,+    sendMessageWithHeadersForLater,+  )+import Pgmq.Effectful.Interpreter (PgmqError (..), runPgmq)+import Pgmq.Effectful.Interpreter.Traced+  ( TracingConfig (..),+    defaultTracingConfig,+    runPgmqTraced,+    runPgmqTracedWith,+  )+import Pgmq.Effectful.Telemetry+  ( TraceHeaders,+    extractTraceContext,+    injectTraceContext,+    mergeTraceHeaders,+  )+import Pgmq.Effectful.Traced+  ( MessageWithContext,+    readMessageWithContext,+    sendMessageTraced,+  )+import Pgmq.Hasql.Statements.Types+  ( BatchMessageQuery (..),+    BatchSendMessage (..),+    BatchSendMessageForLater (..),+    BatchSendMessageWithHeaders (..),+    BatchSendMessageWithHeadersForLater (..),+    BatchVisibilityTimeoutQuery (..),+    CreatePartitionedQueue (..),+    EnableNotifyInsert (..),+    MessageQuery (..),+    PopMessage (..),+    QueueMetrics (..),+    ReadMessage (..),+    ReadWithPollMessage (..),+    SendMessage (..),+    SendMessageForLater (..),+    SendMessageWithHeaders (..),+    SendMessageWithHeadersForLater (..),+    VisibilityTimeoutQuery (..),+  )+import Pgmq.Types+  ( Message (..),+    MessageBody (..),+    MessageHeaders (..),+    MessageId (..),+    Queue (..),+    QueueName,+    parseQueueName,+    queueNameToText,+  )
+ src/Pgmq/Effectful/Effect.hs view
@@ -0,0 +1,261 @@+module Pgmq.Effectful.Effect+  ( -- * Effect+    Pgmq (..),++    -- * Queue Management+    createQueue,+    dropQueue,+    createPartitionedQueue,+    createUnloggedQueue,+    detachArchive,++    -- ** Notifications (pgmq 1.7.0+)+    enableNotifyInsert,+    disableNotifyInsert,++    -- ** FIFO Index (pgmq 1.8.0+)+    createFifoIndex,+    createFifoIndexesAll,++    -- * Message Operations+    sendMessage,+    sendMessageForLater,+    batchSendMessage,+    batchSendMessageForLater,++    -- ** With Headers (pgmq 1.5.0+)+    sendMessageWithHeaders,+    sendMessageWithHeadersForLater,+    batchSendMessageWithHeaders,+    batchSendMessageWithHeadersForLater,+    readMessage,+    deleteMessage,+    batchDeleteMessages,+    archiveMessage,+    batchArchiveMessages,+    deleteAllMessagesFromQueue,+    changeVisibilityTimeout,+    batchChangeVisibilityTimeout,++    -- ** Timestamp-based VT (pgmq 1.10.0+)+    setVisibilityTimeoutAt,+    batchSetVisibilityTimeoutAt,+    readWithPoll,+    pop,++    -- ** FIFO Read (pgmq 1.8.0+)+    readGrouped,+    readGroupedWithPoll,++    -- ** Round-Robin FIFO Read (pgmq 1.9.0+)+    readGroupedRoundRobin,+    readGroupedRoundRobinWithPoll,++    -- * Queue Observability+    listQueues,+    queueMetrics,+    allQueueMetrics,+  )+where++import Data.Int (Int64)+import Data.Vector (Vector)+import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>))+import Effectful.Dispatch.Dynamic (send)+import Pgmq.Hasql.Statements.Types+  ( BatchMessageQuery,+    BatchSendMessage,+    BatchSendMessageForLater,+    BatchSendMessageWithHeaders,+    BatchSendMessageWithHeadersForLater,+    BatchVisibilityTimeoutAtQuery,+    BatchVisibilityTimeoutQuery,+    CreatePartitionedQueue,+    EnableNotifyInsert,+    MessageQuery,+    PopMessage,+    QueueMetrics,+    ReadGrouped,+    ReadGroupedWithPoll,+    ReadMessage,+    ReadWithPollMessage,+    SendMessage,+    SendMessageForLater,+    SendMessageWithHeaders,+    SendMessageWithHeadersForLater,+    VisibilityTimeoutAtQuery,+    VisibilityTimeoutQuery,+  )+import Pgmq.Types (Message, MessageId, Queue, QueueName)++-- | Effect for pgmq message queue operations.+data Pgmq :: Effect where+  -- Queue Management+  CreateQueue :: QueueName -> Pgmq m ()+  DropQueue :: QueueName -> Pgmq m Bool+  CreatePartitionedQueue :: CreatePartitionedQueue -> Pgmq m ()+  CreateUnloggedQueue :: QueueName -> Pgmq m ()+  DetachArchive :: QueueName -> Pgmq m ()+  EnableNotifyInsert :: EnableNotifyInsert -> Pgmq m ()+  DisableNotifyInsert :: QueueName -> Pgmq m ()+  CreateFifoIndex :: QueueName -> Pgmq m ()+  CreateFifoIndexesAll :: Pgmq m ()+  -- Message Operations+  SendMessage :: SendMessage -> Pgmq m MessageId+  SendMessageForLater :: SendMessageForLater -> Pgmq m MessageId+  BatchSendMessage :: BatchSendMessage -> Pgmq m [MessageId]+  BatchSendMessageForLater :: BatchSendMessageForLater -> Pgmq m [MessageId]+  SendMessageWithHeaders :: SendMessageWithHeaders -> Pgmq m MessageId+  SendMessageWithHeadersForLater :: SendMessageWithHeadersForLater -> Pgmq m MessageId+  BatchSendMessageWithHeaders :: BatchSendMessageWithHeaders -> Pgmq m [MessageId]+  BatchSendMessageWithHeadersForLater :: BatchSendMessageWithHeadersForLater -> Pgmq m [MessageId]+  ReadMessage :: ReadMessage -> Pgmq m (Vector Message)+  DeleteMessage :: MessageQuery -> Pgmq m Bool+  BatchDeleteMessages :: BatchMessageQuery -> Pgmq m [MessageId]+  ArchiveMessage :: MessageQuery -> Pgmq m Bool+  BatchArchiveMessages :: BatchMessageQuery -> Pgmq m [MessageId]+  DeleteAllMessagesFromQueue :: QueueName -> Pgmq m Int64+  ChangeVisibilityTimeout :: VisibilityTimeoutQuery -> Pgmq m Message+  BatchChangeVisibilityTimeout :: BatchVisibilityTimeoutQuery -> Pgmq m (Vector Message)+  -- Timestamp-based VT (pgmq 1.10.0+)+  SetVisibilityTimeoutAt :: VisibilityTimeoutAtQuery -> Pgmq m Message+  BatchSetVisibilityTimeoutAt :: BatchVisibilityTimeoutAtQuery -> Pgmq m (Vector Message)+  ReadWithPoll :: ReadWithPollMessage -> Pgmq m (Vector Message)+  Pop :: PopMessage -> Pgmq m (Vector Message)+  -- FIFO Read (pgmq 1.8.0+)+  ReadGrouped :: ReadGrouped -> Pgmq m (Vector Message)+  ReadGroupedWithPoll :: ReadGroupedWithPoll -> Pgmq m (Vector Message)+  -- Round-robin FIFO Read (pgmq 1.9.0+)+  ReadGroupedRoundRobin :: ReadGrouped -> Pgmq m (Vector Message)+  ReadGroupedRoundRobinWithPoll :: ReadGroupedWithPoll -> Pgmq m (Vector Message)+  -- Queue Observability+  ListQueues :: Pgmq m [Queue]+  QueueMetrics :: QueueName -> Pgmq m QueueMetrics+  AllQueueMetrics :: Pgmq m [QueueMetrics]++type instance DispatchOf Pgmq = 'Dynamic++-- Queue Management++createQueue :: (Pgmq :> es) => QueueName -> Eff es ()+createQueue = send . CreateQueue++dropQueue :: (Pgmq :> es) => QueueName -> Eff es Bool+dropQueue = send . DropQueue++createPartitionedQueue :: (Pgmq :> es) => CreatePartitionedQueue -> Eff es ()+createPartitionedQueue = send . CreatePartitionedQueue++createUnloggedQueue :: (Pgmq :> es) => QueueName -> Eff es ()+createUnloggedQueue = send . CreateUnloggedQueue++{-# DEPRECATED detachArchive "detach_archive is a no-op in pgmq and will be removed in pgmq 2.0" #-}+detachArchive :: (Pgmq :> es) => QueueName -> Eff es ()+detachArchive = send . DetachArchive++enableNotifyInsert :: (Pgmq :> es) => EnableNotifyInsert -> Eff es ()+enableNotifyInsert = send . EnableNotifyInsert++disableNotifyInsert :: (Pgmq :> es) => QueueName -> Eff es ()+disableNotifyInsert = send . DisableNotifyInsert++-- | Create FIFO index for a queue (pgmq 1.8.0+)+createFifoIndex :: (Pgmq :> es) => QueueName -> Eff es ()+createFifoIndex = send . CreateFifoIndex++-- | Create FIFO indexes for all queues (pgmq 1.8.0+)+createFifoIndexesAll :: (Pgmq :> es) => Eff es ()+createFifoIndexesAll = send CreateFifoIndexesAll++-- Message Operations++sendMessage :: (Pgmq :> es) => SendMessage -> Eff es MessageId+sendMessage = send . SendMessage++sendMessageForLater :: (Pgmq :> es) => SendMessageForLater -> Eff es MessageId+sendMessageForLater = send . SendMessageForLater++batchSendMessage :: (Pgmq :> es) => BatchSendMessage -> Eff es [MessageId]+batchSendMessage = send . BatchSendMessage++batchSendMessageForLater :: (Pgmq :> es) => BatchSendMessageForLater -> Eff es [MessageId]+batchSendMessageForLater = send . BatchSendMessageForLater++sendMessageWithHeaders :: (Pgmq :> es) => SendMessageWithHeaders -> Eff es MessageId+sendMessageWithHeaders = send . SendMessageWithHeaders++sendMessageWithHeadersForLater :: (Pgmq :> es) => SendMessageWithHeadersForLater -> Eff es MessageId+sendMessageWithHeadersForLater = send . SendMessageWithHeadersForLater++batchSendMessageWithHeaders :: (Pgmq :> es) => BatchSendMessageWithHeaders -> Eff es [MessageId]+batchSendMessageWithHeaders = send . BatchSendMessageWithHeaders++batchSendMessageWithHeadersForLater :: (Pgmq :> es) => BatchSendMessageWithHeadersForLater -> Eff es [MessageId]+batchSendMessageWithHeadersForLater = send . BatchSendMessageWithHeadersForLater++readMessage :: (Pgmq :> es) => ReadMessage -> Eff es (Vector Message)+readMessage = send . ReadMessage++deleteMessage :: (Pgmq :> es) => MessageQuery -> Eff es Bool+deleteMessage = send . DeleteMessage++batchDeleteMessages :: (Pgmq :> es) => BatchMessageQuery -> Eff es [MessageId]+batchDeleteMessages = send . BatchDeleteMessages++archiveMessage :: (Pgmq :> es) => MessageQuery -> Eff es Bool+archiveMessage = send . ArchiveMessage++batchArchiveMessages :: (Pgmq :> es) => BatchMessageQuery -> Eff es [MessageId]+batchArchiveMessages = send . BatchArchiveMessages++deleteAllMessagesFromQueue :: (Pgmq :> es) => QueueName -> Eff es Int64+deleteAllMessagesFromQueue = send . DeleteAllMessagesFromQueue++changeVisibilityTimeout :: (Pgmq :> es) => VisibilityTimeoutQuery -> Eff es Message+changeVisibilityTimeout = send . ChangeVisibilityTimeout++batchChangeVisibilityTimeout :: (Pgmq :> es) => BatchVisibilityTimeoutQuery -> Eff es (Vector Message)+batchChangeVisibilityTimeout = send . BatchChangeVisibilityTimeout++-- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+)+setVisibilityTimeoutAt :: (Pgmq :> es) => VisibilityTimeoutAtQuery -> Eff es Message+setVisibilityTimeoutAt = send . SetVisibilityTimeoutAt++-- | Batch set visibility timeout to an absolute timestamp (pgmq 1.10.0+)+batchSetVisibilityTimeoutAt :: (Pgmq :> es) => BatchVisibilityTimeoutAtQuery -> Eff es (Vector Message)+batchSetVisibilityTimeoutAt = send . BatchSetVisibilityTimeoutAt++readWithPoll :: (Pgmq :> es) => ReadWithPollMessage -> Eff es (Vector Message)+readWithPoll = send . ReadWithPoll++pop :: (Pgmq :> es) => PopMessage -> Eff es (Vector Message)+pop = send . Pop++-- FIFO Read (pgmq 1.8.0+)++-- | FIFO read - fills batch from same message group (pgmq 1.8.0+)+readGrouped :: (Pgmq :> es) => ReadGrouped -> Eff es (Vector Message)+readGrouped = send . ReadGrouped++-- | FIFO read with polling (pgmq 1.8.0+)+readGroupedWithPoll :: (Pgmq :> es) => ReadGroupedWithPoll -> Eff es (Vector Message)+readGroupedWithPoll = send . ReadGroupedWithPoll++-- | Round-robin FIFO read (pgmq 1.9.0+)+readGroupedRoundRobin :: (Pgmq :> es) => ReadGrouped -> Eff es (Vector Message)+readGroupedRoundRobin = send . ReadGroupedRoundRobin++-- | Round-robin FIFO read with polling (pgmq 1.9.0+)+readGroupedRoundRobinWithPoll :: (Pgmq :> es) => ReadGroupedWithPoll -> Eff es (Vector Message)+readGroupedRoundRobinWithPoll = send . ReadGroupedRoundRobinWithPoll++-- Queue Observability++listQueues :: (Pgmq :> es) => Eff es [Queue]+listQueues = send ListQueues++queueMetrics :: (Pgmq :> es) => QueueName -> Eff es QueueMetrics+queueMetrics = send . QueueMetrics++allQueueMetrics :: (Pgmq :> es) => Eff es [QueueMetrics]+allQueueMetrics = send AllQueueMetrics
+ src/Pgmq/Effectful/Interpreter.hs view
@@ -0,0 +1,85 @@+module Pgmq.Effectful.Interpreter+  ( -- * Interpreters+    runPgmq,++    -- * Error Types+    PgmqError (..),+  )+where++import Effectful (Eff, IOE, (:>))+import Effectful qualified+import Effectful.Dispatch.Dynamic (interpret)+import Effectful.Error.Static (Error, throwError)+import Hasql.Pool (Pool, UsageError)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified+import Pgmq.Effectful.Effect (Pgmq (..))+import Pgmq.Hasql.Sessions qualified as Sessions++-- | Error type for pgmq operations.+newtype PgmqError = PgmqPoolError UsageError+  deriving stock (Show)++-- | Run the Pgmq effect using a connection pool.+-- Errors are thrown via the 'Error' effect.+runPgmq ::+  (IOE :> es, Error PgmqError :> es) =>+  Pool ->+  Eff (Pgmq : es) a ->+  Eff es a+runPgmq pool = interpret $ \_ -> \case+  -- Queue Management+  CreateQueue q -> runSession pool $ Sessions.createQueue q+  DropQueue q -> runSession pool $ Sessions.dropQueue q+  CreatePartitionedQueue q -> runSession pool $ Sessions.createPartitionedQueue q+  CreateUnloggedQueue q -> runSession pool $ Sessions.createUnloggedQueue q+  DetachArchive _q -> pure ()+  EnableNotifyInsert config -> runSession pool $ Sessions.enableNotifyInsert config+  DisableNotifyInsert q -> runSession pool $ Sessions.disableNotifyInsert q+  CreateFifoIndex q -> runSession pool $ Sessions.createFifoIndex q+  CreateFifoIndexesAll -> runSession pool Sessions.createFifoIndexesAll+  -- Message Operations+  SendMessage msg -> runSession pool $ Sessions.sendMessage msg+  SendMessageForLater msg -> runSession pool $ Sessions.sendMessageForLater msg+  BatchSendMessage msgs -> runSession pool $ Sessions.batchSendMessage msgs+  BatchSendMessageForLater msgs -> runSession pool $ Sessions.batchSendMessageForLater msgs+  SendMessageWithHeaders msg -> runSession pool $ Sessions.sendMessageWithHeaders msg+  SendMessageWithHeadersForLater msg -> runSession pool $ Sessions.sendMessageWithHeadersForLater msg+  BatchSendMessageWithHeaders msgs -> runSession pool $ Sessions.batchSendMessageWithHeaders msgs+  BatchSendMessageWithHeadersForLater msgs -> runSession pool $ Sessions.batchSendMessageWithHeadersForLater msgs+  ReadMessage query -> runSession pool $ Sessions.readMessage query+  DeleteMessage query -> runSession pool $ Sessions.deleteMessage query+  BatchDeleteMessages query -> runSession pool $ Sessions.batchDeleteMessages query+  ArchiveMessage query -> runSession pool $ Sessions.archiveMessage query+  BatchArchiveMessages query -> runSession pool $ Sessions.batchArchiveMessages query+  DeleteAllMessagesFromQueue q -> runSession pool $ Sessions.deleteAllMessagesFromQueue q+  ChangeVisibilityTimeout query -> runSession pool $ Sessions.changeVisibilityTimeout query+  BatchChangeVisibilityTimeout query -> runSession pool $ Sessions.batchChangeVisibilityTimeout query+  -- Timestamp-based VT (pgmq 1.10.0+)+  SetVisibilityTimeoutAt query -> runSession pool $ Sessions.setVisibilityTimeoutAt query+  BatchSetVisibilityTimeoutAt query -> runSession pool $ Sessions.batchSetVisibilityTimeoutAt query+  ReadWithPoll query -> runSession pool $ Sessions.readWithPoll query+  Pop query -> runSession pool $ Sessions.pop query+  -- FIFO Read (pgmq 1.8.0+)+  ReadGrouped query -> runSession pool $ Sessions.readGrouped query+  ReadGroupedWithPoll query -> runSession pool $ Sessions.readGroupedWithPoll query+  -- Round-robin FIFO Read (pgmq 1.9.0+)+  ReadGroupedRoundRobin query -> runSession pool $ Sessions.readGroupedRoundRobin query+  ReadGroupedRoundRobinWithPoll query -> runSession pool $ Sessions.readGroupedRoundRobinWithPoll query+  -- Queue Observability+  ListQueues -> runSession pool Sessions.listQueues+  QueueMetrics q -> runSession pool $ Sessions.queueMetrics q+  AllQueueMetrics -> runSession pool Sessions.allQueueMetrics++-- Internal helper+runSession ::+  (IOE :> es, Error PgmqError :> es) =>+  Pool ->+  Hasql.Session.Session a ->+  Eff es a+runSession pool session = do+  result <- Effectful.liftIO $ Pool.use pool session+  case result of+    Left err -> throwError $ PgmqPoolError err+    Right a -> pure a
+ src/Pgmq/Effectful/Interpreter/Traced.hs view
@@ -0,0 +1,323 @@+-- | OpenTelemetry-instrumented interpreter for the Pgmq effect.+--+-- This module provides a traced version of the Pgmq interpreter that+-- creates OpenTelemetry spans for all PGMQ operations.+--+-- == Span Kinds+--+-- * Producer spans: Queue creation, message send operations+-- * Consumer spans: Message read operations+-- * Internal spans: Message lifecycle (delete, archive, visibility timeout)+--+-- == Usage+--+-- @+-- import OpenTelemetry.Trace qualified as OTel+-- import Pgmq.Effectful.Interpreter.Traced+--+-- main :: IO ()+-- main = do+--   tracerProvider <- OTel.getGlobalTracerProvider+--   let tracer = OTel.makeTracer tracerProvider "my-app" OTel.tracerOptions+--   pool <- ...+--+--   runEff+--     . runError @PgmqError+--     . runPgmqTraced pool tracer+--     $ do+--       createQueue "my-queue"+--       sendMessage $ SendMessage "my-queue" (MessageBody "hello") Nothing+-- @+module Pgmq.Effectful.Interpreter.Traced+  ( -- * Interpreters+    runPgmqTraced,+    runPgmqTracedWith,++    -- * Configuration+    TracingConfig (..),+    defaultTracingConfig,+  )+where++import Control.Monad (when)+import Data.Text (Text)+import Data.Text qualified as T+import Effectful (Eff, IOE, (:>))+import Effectful qualified+import Effectful.Dispatch.Dynamic (interpret)+import Hasql.Pool (Pool, UsageError)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified+import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..))+import OpenTelemetry.Attributes.Map qualified as AttrMap+import OpenTelemetry.Trace.Core qualified as OTel+import Pgmq.Effectful.Effect (Pgmq (..))+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)++-- | Configuration for OpenTelemetry tracing.+data TracingConfig = TracingConfig+  { -- | The OpenTelemetry tracer to use+    tracer :: !OTel.Tracer,+    -- | Whether to record exceptions on spans+    recordExceptions :: !Bool,+    -- | Whether to include message bodies in spans (may contain PII)+    includeMessageBodies :: !Bool+  }++-- | Create a default tracing configuration.+defaultTracingConfig :: OTel.Tracer -> TracingConfig+defaultTracingConfig t =+  TracingConfig+    { tracer = t,+      recordExceptions = True,+      includeMessageBodies = False+    }++-- | Run the Pgmq effect with OpenTelemetry instrumentation.+runPgmqTraced ::+  (IOE :> es) =>+  Pool ->+  OTel.Tracer ->+  Eff (Pgmq : es) a ->+  Eff es a+runPgmqTraced pool t = runPgmqTracedWith pool (defaultTracingConfig t)++-- | Run the Pgmq effect with custom tracing configuration.+runPgmqTracedWith ::+  (IOE :> es) =>+  Pool ->+  TracingConfig ->+  Eff (Pgmq : es) a ->+  Eff es a+runPgmqTracedWith pool config = interpret $ \_ -> \case+  -- Queue Management (Producer spans)+  CreateQueue q ->+    withTracedSession config "pgmq create_queue" OTel.Producer q pool $+      Sessions.createQueue q+  DropQueue q ->+    withTracedSession config "pgmq drop_queue" OTel.Producer q pool $+      Sessions.dropQueue q+  CreatePartitionedQueue pq@(Types.CreatePartitionedQueue qn _ _) ->+    withTracedSession config "pgmq create_partitioned_queue" OTel.Producer qn pool $+      Sessions.createPartitionedQueue pq+  CreateUnloggedQueue q ->+    withTracedSession config "pgmq create_unlogged_queue" OTel.Producer q pool $+      Sessions.createUnloggedQueue q+  DetachArchive _q ->+    pure ()+  EnableNotifyInsert cfg@(Types.EnableNotifyInsert qn _) ->+    withTracedSession config "pgmq enable_notify_insert" OTel.Internal qn pool $+      Sessions.enableNotifyInsert cfg+  DisableNotifyInsert q ->+    withTracedSession config "pgmq disable_notify_insert" OTel.Internal q pool $+      Sessions.disableNotifyInsert q+  CreateFifoIndex q ->+    withTracedSession config "pgmq create_fifo_index" OTel.Internal q pool $+      Sessions.createFifoIndex q+  CreateFifoIndexesAll ->+    withTracedSessionNoQueue config "pgmq create_fifo_indexes_all" OTel.Internal pool $+      Sessions.createFifoIndexesAll+  -- Message Operations - Send (Producer spans)+  SendMessage msg@(Types.SendMessage qn _ _) ->+    withTracedSession config "pgmq send" OTel.Producer qn pool $+      Sessions.sendMessage msg+  SendMessageForLater msg@(Types.SendMessageForLater qn _ _) ->+    withTracedSession config "pgmq send" OTel.Producer qn pool $+      Sessions.sendMessageForLater msg+  BatchSendMessage msg@(Types.BatchSendMessage qn bodies _) ->+    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+      Sessions.batchSendMessage msg+  BatchSendMessageForLater msg@(Types.BatchSendMessageForLater qn bodies _) ->+    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+      Sessions.batchSendMessageForLater msg+  SendMessageWithHeaders msg@(Types.SendMessageWithHeaders qn _ _ _) ->+    withTracedSession config "pgmq send" OTel.Producer qn pool $+      Sessions.sendMessageWithHeaders msg+  SendMessageWithHeadersForLater msg@(Types.SendMessageWithHeadersForLater qn _ _ _) ->+    withTracedSession config "pgmq send" OTel.Producer qn pool $+      Sessions.sendMessageWithHeadersForLater msg+  BatchSendMessageWithHeaders msg@(Types.BatchSendMessageWithHeaders qn bodies _ _) ->+    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+      Sessions.batchSendMessageWithHeaders msg+  BatchSendMessageWithHeadersForLater msg@(Types.BatchSendMessageWithHeadersForLater qn bodies _ _) ->+    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+      Sessions.batchSendMessageWithHeadersForLater msg+  -- Message Operations - Read (Consumer spans)+  ReadMessage query@(Types.ReadMessage qn _ _ _) ->+    withTracedSession config "pgmq read" OTel.Consumer qn pool $+      Sessions.readMessage query+  ReadWithPoll query@(Types.ReadWithPollMessage qn _ _ _ _ _) ->+    withTracedSession config "pgmq read_with_poll" OTel.Consumer qn pool $+      Sessions.readWithPoll query+  Pop query@(Types.PopMessage qn _) ->+    withTracedSession config "pgmq pop" OTel.Consumer qn pool $+      Sessions.pop query+  -- FIFO Read (Consumer spans)+  ReadGrouped query@(Types.ReadGrouped qn _ _) ->+    withTracedSession config "pgmq read_grouped" OTel.Consumer qn pool $+      Sessions.readGrouped query+  ReadGroupedWithPoll query@(Types.ReadGroupedWithPoll qn _ _ _ _) ->+    withTracedSession config "pgmq read_grouped_with_poll" OTel.Consumer qn pool $+      Sessions.readGroupedWithPoll query+  -- Round-robin FIFO Read (Consumer spans)+  ReadGroupedRoundRobin query@(Types.ReadGrouped qn _ _) ->+    withTracedSession config "pgmq read_grouped_rr" OTel.Consumer qn pool $+      Sessions.readGroupedRoundRobin query+  ReadGroupedRoundRobinWithPoll query@(Types.ReadGroupedWithPoll qn _ _ _ _) ->+    withTracedSession config "pgmq read_grouped_rr_with_poll" OTel.Consumer qn pool $+      Sessions.readGroupedRoundRobinWithPoll query+  -- Message Lifecycle (Internal spans)+  DeleteMessage query@(Types.MessageQuery qn msgId) ->+    withTracedSessionWithMsgId config "pgmq delete" OTel.Internal qn msgId pool $+      Sessions.deleteMessage query+  BatchDeleteMessages query@(Types.BatchMessageQuery qn msgIds) ->+    withTracedSessionWithCount config "pgmq delete_batch" OTel.Internal qn (length msgIds) pool $+      Sessions.batchDeleteMessages query+  ArchiveMessage query@(Types.MessageQuery qn msgId) ->+    withTracedSessionWithMsgId config "pgmq archive" OTel.Internal qn msgId pool $+      Sessions.archiveMessage query+  BatchArchiveMessages query@(Types.BatchMessageQuery qn msgIds) ->+    withTracedSessionWithCount config "pgmq archive_batch" OTel.Internal qn (length msgIds) pool $+      Sessions.batchArchiveMessages query+  DeleteAllMessagesFromQueue q ->+    withTracedSession config "pgmq purge_queue" OTel.Internal q pool $+      Sessions.deleteAllMessagesFromQueue q+  ChangeVisibilityTimeout query@(Types.VisibilityTimeoutQuery qn msgId _) ->+    withTracedSessionWithMsgId config "pgmq set_vt" OTel.Internal qn msgId pool $+      Sessions.changeVisibilityTimeout query+  BatchChangeVisibilityTimeout query@(Types.BatchVisibilityTimeoutQuery qn msgIds _) ->+    withTracedSessionWithCount config "pgmq set_vt_batch" OTel.Internal qn (length msgIds) pool $+      Sessions.batchChangeVisibilityTimeout query+  -- Timestamp-based VT (pgmq 1.10.0+)+  SetVisibilityTimeoutAt query@(Types.VisibilityTimeoutAtQuery qn msgId _) ->+    withTracedSessionWithMsgId config "pgmq set_vt_at" OTel.Internal qn msgId pool $+      Sessions.setVisibilityTimeoutAt query+  BatchSetVisibilityTimeoutAt query@(Types.BatchVisibilityTimeoutAtQuery qn msgIds _) ->+    withTracedSessionWithCount config "pgmq set_vt_at_batch" OTel.Internal qn (length msgIds) pool $+      Sessions.batchSetVisibilityTimeoutAt query+  -- Queue Observability (Internal spans)+  ListQueues ->+    withTracedSessionNoQueue config "pgmq list_queues" OTel.Internal pool $+      Sessions.listQueues+  QueueMetrics q ->+    withTracedSession config "pgmq metrics" OTel.Internal q pool $+      Sessions.queueMetrics q+  AllQueueMetrics ->+    withTracedSessionNoQueue config "pgmq metrics_all" OTel.Internal pool $+      Sessions.allQueueMetrics++-- Internal helpers++withTracedSession ::+  (IOE :> es) =>+  TracingConfig ->+  Text ->+  OTel.SpanKind ->+  QueueName ->+  Pool ->+  Hasql.Session.Session a ->+  Eff es a+withTracedSession config spanName kind queueName pool session = do+  let args = OTel.defaultSpanArguments {OTel.kind = kind}+  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do+    addQueueAttributes s queueName+    runSessionIO config s pool session++withTracedSessionNoQueue ::+  (IOE :> es) =>+  TracingConfig ->+  Text ->+  OTel.SpanKind ->+  Pool ->+  Hasql.Session.Session a ->+  Eff es a+withTracedSessionNoQueue config spanName kind pool session = do+  let args = OTel.defaultSpanArguments {OTel.kind = kind}+  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do+    addBaseAttributes s+    runSessionIO config s pool session++withTracedSessionWithMsgId ::+  (IOE :> es) =>+  TracingConfig ->+  Text ->+  OTel.SpanKind ->+  QueueName ->+  MessageId ->+  Pool ->+  Hasql.Session.Session a ->+  Eff es a+withTracedSessionWithMsgId config spanName kind queueName msgId pool session = do+  let args = OTel.defaultSpanArguments {OTel.kind = kind}+  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do+    addQueueAttributes s queueName+    addMessageIdAttribute s msgId+    runSessionIO config s pool session++withTracedSessionWithCount ::+  (IOE :> es) =>+  TracingConfig ->+  Text ->+  OTel.SpanKind ->+  QueueName ->+  Int ->+  Pool ->+  Hasql.Session.Session a ->+  Eff es a+withTracedSessionWithCount config spanName kind queueName count pool session = do+  let args = OTel.defaultSpanArguments {OTel.kind = kind}+  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do+    addQueueAttributes s queueName+    addBatchCountAttribute s count+    runSessionIO config s pool session++runSessionIO ::+  TracingConfig ->+  OTel.Span ->+  Pool ->+  Hasql.Session.Session a ->+  IO a+runSessionIO config s pool session = do+  result <- Pool.use pool session+  case result of+    Left err -> do+      when config.recordExceptions $ recordUsageError s err+      OTel.setStatus s (OTel.Error $ T.pack $ show err)+      fail $ "PgmqPoolError: " <> show err+    Right a -> do+      OTel.setStatus s OTel.Ok+      pure a++recordUsageError :: OTel.Span -> UsageError -> IO ()+recordUsageError s err = do+  OTel.addEvent s $+    OTel.NewEvent+      { OTel.newEventName = "exception",+        OTel.newEventTimestamp = Nothing,+        OTel.newEventAttributes =+          AttrMap.fromList+            [ ("exception.type", AttributeValue $ TextAttribute "UsageError"),+              ("exception.message", AttributeValue $ TextAttribute $ T.pack $ show err)+            ]+      }++addQueueAttributes :: OTel.Span -> QueueName -> IO ()+addQueueAttributes s queueName = do+  addBaseAttributes s+  OTel.addAttribute s messagingDestinationName (queueNameToText queueName)++addBaseAttributes :: OTel.Span -> IO ()+addBaseAttributes s = do+  OTel.addAttribute s messagingSystem ("pgmq" :: Text)+  OTel.addAttribute s dbSystem ("postgresql" :: Text)++addMessageIdAttribute :: OTel.Span -> MessageId -> IO ()+addMessageIdAttribute s (MessageId msgId) =+  OTel.addAttribute s messagingMessageId (T.pack $ show msgId)++addBatchCountAttribute :: OTel.Span -> Int -> IO ()+addBatchCountAttribute s count =+  OTel.addAttribute s messagingBatchMessageCount count
+ src/Pgmq/Effectful/Telemetry.hs view
@@ -0,0 +1,110 @@+-- | OpenTelemetry utilities for PGMQ tracing.+--+-- This module provides:+--+-- * W3C Trace Context propagation via message headers+-- * OpenTelemetry semantic conventions for messaging systems+module Pgmq.Effectful.Telemetry+  ( -- * Trace Context Operations+    injectTraceContext,+    extractTraceContext,+    mergeTraceHeaders,+    TraceHeaders,++    -- * Semantic Conventions+    messagingSystem,+    messagingOperationType,+    messagingDestinationName,+    messagingMessageId,+    messagingBatchMessageCount,+    dbSystem,+    dbOperationName,+  )+where++import Data.Aeson (Value (..))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KM+import Data.ByteString (ByteString)+import Data.Text (Text)+import Data.Text.Encoding qualified as TE+import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C+import OpenTelemetry.Trace.Core qualified as OTel++-- | Headers for W3C Trace Context propagation.+-- Contains traceparent and optionally tracestate.+type TraceHeaders = [(ByteString, ByteString)]++-- | Inject trace context into message headers (for producers).+-- Creates traceparent and tracestate headers from current span context.+injectTraceContext :: OTel.Span -> IO TraceHeaders+injectTraceContext otelSpan = do+  (traceparent, tracestate) <- W3C.encodeSpanContext otelSpan+  pure+    [ ("traceparent", traceparent),+      ("tracestate", tracestate)+    ]++-- | Extract trace context from message headers (for consumers).+-- Returns SpanContext that can be used as parent link.+extractTraceContext :: TraceHeaders -> Maybe OTel.SpanContext+extractTraceContext headers = do+  let traceparent = lookup "traceparent" headers+      tracestate = lookup "tracestate" headers+  W3C.decodeSpanContext traceparent tracestate++-- | Merge trace headers into existing PGMQ message headers.+-- Preserves existing headers while adding trace context.+mergeTraceHeaders :: TraceHeaders -> Maybe Value -> Value+mergeTraceHeaders traceHeaders existingHeaders =+  let traceObj =+        Object $+          KM.fromList+            [ (Key.fromText "traceparent", toJsonValue $ lookup "traceparent" traceHeaders),+              (Key.fromText "tracestate", toJsonValue $ lookup "tracestate" traceHeaders)+            ]+   in case existingHeaders of+        Just (Object obj) -> Object $ KM.union (toKeyMap traceObj) obj+        Just v -> Object $ KM.fromList [(Key.fromText "_original", v), (Key.fromText "_trace", traceObj)]+        Nothing -> traceObj+  where+    toJsonValue :: Maybe ByteString -> Value+    toJsonValue = maybe Null (String . TE.decodeUtf8)++    toKeyMap :: Value -> KM.KeyMap Value+    toKeyMap (Object o) = o+    toKeyMap _ = KM.empty++-- OpenTelemetry Semantic Conventions for Messaging+-- See: https://opentelemetry.io/docs/specs/semconv/messaging/++-- | The messaging system identifier.+-- Value: "pgmq"+messagingSystem :: Text+messagingSystem = "messaging.system"++-- | The type of messaging operation.+-- Values: "send", "receive", "process"+messagingOperationType :: Text+messagingOperationType = "messaging.operation.type"++-- | The destination queue name.+messagingDestinationName :: Text+messagingDestinationName = "messaging.destination.name"++-- | The unique message identifier.+messagingMessageId :: Text+messagingMessageId = "messaging.message.id"++-- | Number of messages in a batch operation.+messagingBatchMessageCount :: Text+messagingBatchMessageCount = "messaging.batch.message_count"++-- | The database system.+-- Value: "postgresql"+dbSystem :: Text+dbSystem = "db.system"++-- | The database operation name.+dbOperationName :: Text+dbOperationName = "db.operation.name"
+ src/Pgmq/Effectful/Traced.hs view
@@ -0,0 +1,126 @@+-- | Traced operations for PGMQ with automatic trace context propagation.+--+-- This module provides higher-level operations that automatically inject+-- W3C Trace Context into message headers (for producers) and extract+-- trace context from received messages (for consumers).+--+-- == Usage+--+-- @+-- -- Producer: Send with trace context+-- sendMessageTraced tracer queueName body Nothing+--+-- -- Consumer: Read with trace context extraction+-- messagesWithCtx <- readMessageWithContext readQuery+-- forM_ messagesWithCtx $ \(msg, maybeParentCtx) ->+--   -- maybeParentCtx contains the SpanContext from the producer+--   processMessage msg maybeParentCtx+-- @+module Pgmq.Effectful.Traced+  ( -- * Traced Send Operations+    sendMessageTraced,++    -- * Context-aware Read Operations+    readMessageWithContext,+    MessageWithContext,+  )+where++import Data.Aeson (Value (..))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KM+import Data.Text (Text)+import Data.Text.Encoding qualified as TE+import Data.Vector (Vector)+import Data.Vector qualified as V+import Effectful (Eff, IOE, (:>))+import Effectful qualified+import OpenTelemetry.Context qualified as Ctxt+import OpenTelemetry.Context.ThreadLocal qualified as CtxtLocal+import OpenTelemetry.Trace.Core qualified as OTel+import Pgmq.Effectful.Effect (Pgmq, readMessage, sendMessageWithHeaders)+import Pgmq.Effectful.Telemetry+import Pgmq.Hasql.Statements.Types qualified as Types+import Pgmq.Types++-- | A message paired with its extracted trace context (if present).+type MessageWithContext = (Message, Maybe OTel.SpanContext)++-- | Send a message with trace context automatically injected into headers.+--+-- This creates traceparent and tracestate headers from the current span+-- context, which can be extracted by consumers to link traces.+--+-- If existing headers are provided, the trace headers are merged in.+sendMessageTraced ::+  (Pgmq :> es, IOE :> es) =>+  OTel.Tracer ->+  QueueName ->+  MessageBody ->+  Maybe Value ->+  Eff es MessageId+sendMessageTraced _tracer queueName body existingHeaders = do+  -- Get current span context and inject into headers+  ctx <- Effectful.liftIO CtxtLocal.getContext+  traceHeaders <- case Ctxt.lookupSpan ctx of+    Just s -> Effectful.liftIO $ injectTraceContext s+    Nothing -> pure []++  let mergedHeaders = MessageHeaders $ mergeTraceHeaders traceHeaders existingHeaders++  sendMessageWithHeaders $+    Types.SendMessageWithHeaders+      { queueName = queueName,+        messageBody = body,+        messageHeaders = mergedHeaders,+        delay = Nothing+      }++-- | Read messages and extract trace context from headers.+--+-- Returns messages paired with their extracted SpanContext (if present).+-- The SpanContext can be used to create a child span that links to the+-- producer's trace.+--+-- @+-- messagesWithCtx <- readMessageWithContext readQuery+-- forM_ messagesWithCtx $ \(msg, maybeParentCtx) ->+--   inSpan' tracer "process" (linkToParent maybeParentCtx) $ do+--     processMessage msg+-- @+readMessageWithContext ::+  (Pgmq :> es) =>+  Types.ReadMessage ->+  Eff es (Vector MessageWithContext)+readMessageWithContext readQuery = do+  messages <- readMessage readQuery+  pure $ V.map extractContext messages+  where+    extractContext :: Message -> MessageWithContext+    extractContext msg =+      let ctx = extractTraceContextFromMessage msg+       in (msg, ctx)++    extractTraceContextFromMessage :: Message -> Maybe OTel.SpanContext+    extractTraceContextFromMessage msg = do+      hdrs <- msg.headers+      traceHeaders <- parseTraceHeaders hdrs+      extractTraceContext traceHeaders++    parseTraceHeaders :: Value -> Maybe TraceHeaders+    parseTraceHeaders (Object obj) = do+      traceparent <- KM.lookup (Key.fromText "traceparent") obj >>= asText+      let tracestate = KM.lookup (Key.fromText "tracestate") obj >>= asText+      pure $+        catMaybes+          [ Just ("traceparent", TE.encodeUtf8 traceparent),+            ("tracestate",) . TE.encodeUtf8 <$> tracestate+          ]+    parseTraceHeaders _ = Nothing++    asText :: Value -> Maybe Text+    asText (String t) = Just t+    asText _ = Nothing++    catMaybes :: [Maybe a] -> [a]+    catMaybes = foldr (\x acc -> maybe acc (: acc) x) []