diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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 diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/pgmq-effectful.cabal b/pgmq-effectful.cabal new file mode 100644 --- /dev/null +++ b/pgmq-effectful.cabal @@ -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 diff --git a/src/Pgmq/Effectful.hs b/src/Pgmq/Effectful.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Effectful.hs @@ -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, + ) diff --git a/src/Pgmq/Effectful/Effect.hs b/src/Pgmq/Effectful/Effect.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Effectful/Effect.hs @@ -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 diff --git a/src/Pgmq/Effectful/Interpreter.hs b/src/Pgmq/Effectful/Interpreter.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Effectful/Interpreter.hs @@ -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 diff --git a/src/Pgmq/Effectful/Interpreter/Traced.hs b/src/Pgmq/Effectful/Interpreter/Traced.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Effectful/Interpreter/Traced.hs @@ -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 diff --git a/src/Pgmq/Effectful/Telemetry.hs b/src/Pgmq/Effectful/Telemetry.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Effectful/Telemetry.hs @@ -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" diff --git a/src/Pgmq/Effectful/Traced.hs b/src/Pgmq/Effectful/Traced.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Effectful/Traced.hs @@ -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) []