pgmq-hasql (empty) → 0.1.0.0

raw patch · 26 files changed

+3339/−0 lines, 26 filesdep +aesondep +basedep +ephemeral-pg

Dependencies added: aeson, base, ephemeral-pg, generic-lens, hasql, hasql-pool, hasql-transaction, hedgehog, lens, pgmq-core, pgmq-hasql, pgmq-migration, random, scientific, tasty, tasty-hedgehog, tasty-hunit, template-haskell, text, time, vector

Files

• CHANGELOG.md +42/−0
• LICENSE +20/−0
• pgmq-hasql.cabal +109/−0
• src/Pgmq.hs +143/−0
• src/Pgmq/Hasql/Decoders.hs +60/−0
• src/Pgmq/Hasql/Encoders.hs +216/−0
• src/Pgmq/Hasql/Prelude.hs +63/−0
• src/Pgmq/Hasql/Quasi.hs +16/−0
• src/Pgmq/Hasql/Sessions.hs +198/−0
• src/Pgmq/Hasql/Statements.hs +10/−0
• src/Pgmq/Hasql/Statements/Message.hs +272/−0
• src/Pgmq/Hasql/Statements/QueueManagement.hs +78/−0
• src/Pgmq/Hasql/Statements/QueueObservability.hs +36/−0
• src/Pgmq/Hasql/Statements/Types.hs +220/−0
• test/AdvancedOpsSpec.hs +378/−0
• test/AllFunctionsDecoderSpec.hs +202/−0
• test/DecoderValidationSpec.hs +282/−0
• test/EphemeralDb.hs +72/−0
• test/Generators.hs +111/−0
• test/Main.hs +33/−0
• test/MessageSpec.hs +300/−0
• test/MetricsSpec.hs +96/−0
• test/QueueSpec.hs +88/−0
• test/RoundTripSpec.hs +144/−0
• test/SchemaSpec.hs +91/−0
• test/TestUtils.hs +59/−0

@@ -0,0 +1,42 @@+# Revision history for pgmq-hasql++## 0.1.0.0 -- 2026-02-21++### New Features++#### pgmq 1.5.0+ Support+- Message headers: `sendMessageWithHeaders`, `sendMessageWithHeadersForLater`,+  `batchSendMessageWithHeaders`, `batchSendMessageWithHeadersForLater`+- Conditional read filtering via `conditional` field in `ReadMessage`+- Added `queueVisibleLength` to `QueueMetrics`++#### pgmq 1.7.0+ Support+- Pop with quantity via `PopMessage` type+- Queue notifications: `enableNotifyInsert`, `disableNotifyInsert`++#### pgmq 1.8.0+ Support+- Batch visibility timeout: `batchChangeVisibilityTimeout`+- Notification throttling via `throttleIntervalMs` in `EnableNotifyInsert`+- FIFO read functions:+  - `readGrouped`: SQS-style batch filling from same message group+  - `readGroupedWithPoll`: Same with polling support+- FIFO index management:+  - `createFifoIndex`: Create FIFO index for a specific queue+  - `createFifoIndexesAll`: Create FIFO indexes for all queues+- New types: `ReadGrouped`, `ReadGroupedWithPoll`++#### pgmq 1.9.0+ Support+- Round-robin FIFO read functions:+  - `readGroupedRoundRobin`: Fair distribution across message groups+  - `readGroupedRoundRobinWithPoll`: Same with polling support+- Note: FIFO functions do not support `conditional` parameter (removed in pgmq 1.9.0)++#### pgmq 1.10.0+ Support+- Timestamp-based `set_vt` API+- `lastReadAt` field on `Message` type++### Deprecations++- `detachArchive` is now deprecated (no-op in pgmq, will be removed in pgmq 2.0)++* Initial release with full PGMQ API coverage

@@ -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.

@@ -0,0 +1,109 @@+cabal-version:   3.4+name:            pgmq-hasql+version:         0.1.0.0+synopsis:        Hasql-based client for PGMQ (PostgreSQL Message Queue)+description:+  A Haskell client library for PGMQ (PostgreSQL Message Queue) built+  on hasql. Provides queue management, message sending\/receiving,+  visibility timeout, archival, FIFO reads, and more. Supports+  pgmq 1.5.0 through 1.10.0 features including message headers,+  conditional reads, notifications, and round-robin FIFO.++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+    Pgmq.Hasql.Decoders+    Pgmq.Hasql.Encoders+    Pgmq.Hasql.Prelude+    Pgmq.Hasql.Quasi+    Pgmq.Hasql.Sessions+    Pgmq.Hasql.Statements+    Pgmq.Hasql.Statements.Message+    Pgmq.Hasql.Statements.QueueManagement+    Pgmq.Hasql.Statements.QueueObservability+    Pgmq.Hasql.Statements.Types++  default-extensions:+    DeriveGeneric+    DuplicateRecordFields+    GeneralisedNewtypeDeriving+    ImportQualifiedPost+    NamedFieldPuns+    OverloadedLabels+    OverloadedStrings++  build-depends:+    , aeson              ^>=2.2+    , base               >=4.18  && <5+    , generic-lens       ^>=2.2  || ^>=2.3+    , hasql              ^>=1.10+    , hasql-transaction  ^>=1.2+    , lens               ^>=5.3+    , pgmq-core          >=0.1   && <0.2+    , template-haskell   >=2.20  && <3+    , text               ^>=2.1+    , time               ^>=1.14+    , vector             ^>=0.13++  hs-source-dirs:     src+  default-language:   GHC2024++test-suite pgmq-hasql-test+  import:             warnings+  default-language:   GHC2024+  type:               exitcode-stdio-1.0+  hs-source-dirs:     test+  main-is:            Main.hs+  ghc-options:        -threaded -rtsopts -with-rtsopts=-N+  other-modules:+    AdvancedOpsSpec+    AllFunctionsDecoderSpec+    DecoderValidationSpec+    EphemeralDb+    Generators+    MessageSpec+    MetricsSpec+    QueueSpec+    RoundTripSpec+    SchemaSpec+    TestUtils++  default-extensions:+    ImportQualifiedPost+    OverloadedStrings++  build-depends:+    , aeson+    , base            >=4.18  && <5+    , ephemeral-pg    >=0.2.1+    , hasql+    , hasql-pool      ^>=1.4+    , hedgehog        ^>=1.5+    , pgmq-core+    , pgmq-hasql+    , pgmq-migration+    , random          ^>=1.2+    , scientific      ^>=0.3+    , tasty           ^>=1.5+    , tasty-hedgehog  ^>=1.4+    , tasty-hunit     ^>=0.10+    , text+    , time+    , vector

@@ -0,0 +1,143 @@+module Pgmq+  ( -- * Queue Management+    createQueue,+    dropQueue,+    createPartitionedQueue,+    createUnloggedQueue,+    detachArchive, -- DEPRECATED: no-op, will be removed in pgmq 2.0++    -- ** 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, -- pgmq 1.8.0+++    -- ** Timestamp-based VT (pgmq 1.10.0+)+    setVisibilityTimeoutAt,+    batchSetVisibilityTimeoutAt,+    listQueues,+    readWithPoll,+    pop,+    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 (..), -- pgmq 1.7.0++    MessageQuery (..),+    BatchMessageQuery (..),+    VisibilityTimeoutQuery (..),+    BatchVisibilityTimeoutQuery (..), -- pgmq 1.8.0+++    -- ** Timestamp-based VT types (pgmq 1.10.0+)+    VisibilityTimeoutAtQuery (..),+    BatchVisibilityTimeoutAtQuery (..),+    ReadWithPollMessage (..),+    CreatePartitionedQueue (..),+    QueueMetrics (..),++    -- * Queue Name Utilities+    parseQueueName,+    queueNameToText,+  )+where++import Pgmq.Hasql.Sessions+  ( allQueueMetrics,+    archiveMessage,+    batchArchiveMessages,+    batchChangeVisibilityTimeout,+    batchDeleteMessages,+    batchSendMessage,+    batchSendMessageForLater,+    batchSendMessageWithHeaders,+    batchSendMessageWithHeadersForLater,+    batchSetVisibilityTimeoutAt,+    changeVisibilityTimeout,+    createPartitionedQueue,+    createQueue,+    createUnloggedQueue,+    deleteAllMessagesFromQueue,+    deleteMessage,+    detachArchive,+    disableNotifyInsert,+    dropQueue,+    enableNotifyInsert,+    listQueues,+    pop,+    queueMetrics,+    readMessage,+    readWithPoll,+    sendMessage,+    sendMessageForLater,+    sendMessageWithHeaders,+    sendMessageWithHeadersForLater,+    setVisibilityTimeoutAt,+  )+import Pgmq.Hasql.Statements.Types+  ( BatchMessageQuery (..),+    BatchSendMessage (..),+    BatchSendMessageForLater (..),+    BatchSendMessageWithHeaders (..),+    BatchSendMessageWithHeadersForLater (..),+    BatchVisibilityTimeoutAtQuery (..),+    BatchVisibilityTimeoutQuery (..),+    CreatePartitionedQueue (..),+    EnableNotifyInsert (..),+    MessageQuery (..),+    PopMessage (..),+    QueueMetrics (..),+    ReadMessage (..),+    ReadWithPollMessage (..),+    SendMessage (..),+    SendMessageForLater (..),+    SendMessageWithHeaders (..),+    SendMessageWithHeadersForLater (..),+    VisibilityTimeoutAtQuery (..),+    VisibilityTimeoutQuery (..),+  )+import Pgmq.Types+  ( Message (..),+    MessageBody (..),+    MessageHeaders (..),+    MessageId (..),+    Queue (..),+    QueueName,+    parseQueueName,+    queueNameToText,+  )

@@ -0,0 +1,60 @@+module Pgmq.Hasql.Decoders+  ( messageDecoder,+    messageIdDecoder,+    queueDecoder,+    queueMetricsDecoder,+  )+where++import Data.Bifunctor (first)+import Data.Text (pack)+import Hasql.Decoders qualified as D+import Pgmq.Hasql.Statements.Types (QueueMetrics (..))+import Pgmq.Types (Message (..), MessageBody (..), MessageId (..), Queue (..), parseQueueName)++-- | Decoder for pgmq.message_record type+-- Column order matches pgmq SQL: msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers+messageDecoder :: D.Row Message+messageDecoder =+  ( \msgId readCt enqueuedAt lastReadAt vt body headers ->+      Message+        { messageId = msgId,+          visibilityTime = vt,+          enqueuedAt = enqueuedAt,+          lastReadAt = lastReadAt,+          readCount = fromIntegral readCt,+          body = body,+          headers = headers+        }+  )+    <$> messageIdDecoder -- msg_id+    <*> D.column (D.nonNullable D.int4) -- read_ct (INTEGER -> Int32)+    <*> D.column (D.nonNullable D.timestamptz) -- enqueued_at+    <*> D.column (D.nullable D.timestamptz) -- last_read_at+    <*> D.column (D.nonNullable D.timestamptz) -- vt+    <*> (MessageBody <$> D.column (D.nonNullable D.jsonb)) -- message+    <*> D.column (D.nullable D.jsonb) -- headers++messageIdDecoder :: D.Row MessageId+messageIdDecoder = MessageId <$> D.column (D.nonNullable D.int8)++-- | Decoder for pgmq.queue_record type+-- Column order: queue_name (varchar), is_partitioned (bool), is_unlogged (bool), created_at (timestamptz)+queueDecoder :: D.Row Queue+queueDecoder =+  (\name isPartitioned isUnlogged createdAt -> Queue name createdAt isPartitioned isUnlogged)+    <$> D.column (D.nonNullable $ D.refine (first (pack . show) . parseQueueName) D.varchar)+    <*> D.column (D.nonNullable D.bool)+    <*> D.column (D.nonNullable D.bool)+    <*> D.column (D.nonNullable D.timestamptz)++queueMetricsDecoder :: D.Row QueueMetrics+queueMetricsDecoder =+  QueueMetrics+    <$> D.column (D.nonNullable D.text)+    <*> D.column (D.nonNullable D.int8)+    <*> D.column (D.nullable D.int4)+    <*> D.column (D.nullable D.int4)+    <*> D.column (D.nonNullable D.int8)+    <*> D.column (D.nonNullable D.timestamptz)+    <*> D.column (D.nonNullable D.int8) -- queue_visible_length (pgmq 1.5.0+)

@@ -0,0 +1,216 @@+module Pgmq.Hasql.Encoders+  ( queueNameValue,+    sendMessageEncoder,+    sendMessageForLaterEncoder,+    batchSendMessageEncoder,+    batchSendMessageForLaterEncoder,+    sendMessageWithHeadersEncoder,+    sendMessageWithHeadersForLaterEncoder,+    batchSendMessageWithHeadersEncoder,+    batchSendMessageWithHeadersForLaterEncoder,+    messageIdValue,+    messageHeadersValue,+    readMessageEncoder,+    popMessageEncoder,+    messageQueryEncoder,+    batchMessageQueryEncoder,+    queueNameEncoder,+    visibilityTimeoutQueryEncoder,+    batchVisibilityTimeoutQueryEncoder,+    -- Timestamp-based VT encoders (pgmq 1.10.0+)+    visibilityTimeoutAtQueryEncoder,+    batchVisibilityTimeoutAtQueryEncoder,+    enableNotifyInsertEncoder,+    readWithPollEncoder,+    createPartitionedQueueEncoder,+    -- FIFO encoders (pgmq 1.8.0+)+    readGroupedEncoder,+    readGroupedWithPollEncoder,+  )+where++import Data.Generics.Product (HasField')+import Hasql.Encoders qualified as E+import Pgmq.Hasql.Prelude+import Pgmq.Hasql.Statements.Types+import Pgmq.Types+  ( MessageBody (..),+    MessageHeaders (..),+    MessageId (..),+    QueueName,+    queueNameToText,+  )++queueNameEncoder :: E.Params QueueName+queueNameEncoder = E.param (E.nonNullable queueNameValue)++queueNameValue :: E.Value QueueName+queueNameValue = queueNameToText >$< E.text++messageBodyValue :: E.Value MessageBody+messageBodyValue = unMessageBody >$< E.jsonb++messageIdValue :: E.Value MessageId+messageIdValue = unMessageId >$< E.int8++messageHeadersValue :: E.Value MessageHeaders+messageHeadersValue = unMessageHeaders >$< E.jsonb++-- | Common encoder for queue message fields+commonSendMessageFields :: (HasField' "queueName" a QueueName, HasField' "messageBody" a MessageBody) => E.Params a+commonSendMessageFields =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageBody >$< E.param (E.nonNullable messageBodyValue))++sendMessageEncoder :: E.Params SendMessage+sendMessageEncoder =+  commonSendMessageFields+    <> (view #delay >$< E.param (E.nullable E.int4))++sendMessageForLaterEncoder :: E.Params SendMessageForLater+sendMessageForLaterEncoder =+  commonSendMessageFields+    <> (view #scheduledAt >$< E.param (E.nonNullable E.timestamptz))++-- | Common encoder for batch message fields+commonBatchSendMessageFields :: (HasField' "queueName" a QueueName, HasField' "messageBodies" a [MessageBody]) => E.Params a+commonBatchSendMessageFields =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageBodies >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageBodyValue))))))++batchSendMessageEncoder :: E.Params BatchSendMessage+batchSendMessageEncoder =+  commonBatchSendMessageFields+    <> (view #delay >$< E.param (E.nullable E.int4))++batchSendMessageForLaterEncoder :: E.Params BatchSendMessageForLater+batchSendMessageForLaterEncoder =+  commonBatchSendMessageFields+    <> (view #scheduledAt >$< E.param (E.nonNullable E.timestamptz))++-- | Encoder for SendMessageWithHeaders (pgmq 1.5.0+)+-- SQL: pgmq.send(queue_name, msg, headers, delay)+sendMessageWithHeadersEncoder :: E.Params SendMessageWithHeaders+sendMessageWithHeadersEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageBody >$< E.param (E.nonNullable messageBodyValue))+    <> (view #messageHeaders >$< E.param (E.nonNullable messageHeadersValue))+    <> (view #delay >$< E.param (E.nullable E.int4))++-- | Encoder for SendMessageWithHeadersForLater (pgmq 1.5.0+)+-- SQL: pgmq.send(queue_name, msg, headers, timestamp)+sendMessageWithHeadersForLaterEncoder :: E.Params SendMessageWithHeadersForLater+sendMessageWithHeadersForLaterEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageBody >$< E.param (E.nonNullable messageBodyValue))+    <> (view #messageHeaders >$< E.param (E.nonNullable messageHeadersValue))+    <> (view #scheduledAt >$< E.param (E.nonNullable E.timestamptz))++-- | Encoder for BatchSendMessageWithHeaders (pgmq 1.5.0+)+-- SQL: pgmq.send_batch(queue_name, msgs[], headers[], delay)+batchSendMessageWithHeadersEncoder :: E.Params BatchSendMessageWithHeaders+batchSendMessageWithHeadersEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageBodies >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageBodyValue))))))+    <> (view #messageHeaders >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageHeadersValue))))))+    <> (view #delay >$< E.param (E.nullable E.int4))++-- | Encoder for BatchSendMessageWithHeadersForLater (pgmq 1.5.0+)+-- SQL: pgmq.send_batch(queue_name, msgs[], headers[], timestamp)+batchSendMessageWithHeadersForLaterEncoder :: E.Params BatchSendMessageWithHeadersForLater+batchSendMessageWithHeadersForLaterEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageBodies >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageBodyValue))))))+    <> (view #messageHeaders >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageHeadersValue))))))+    <> (view #scheduledAt >$< E.param (E.nonNullable E.timestamptz))++-- | Encoder for the 3-param pgmq.read (without conditional filter)+readMessageEncoder :: E.Params ReadMessage+readMessageEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #delay >$< E.param (E.nonNullable E.int4))+    <> (view #batchSize >$< E.param (E.nullable E.int4))++-- | Encoder for PopMessage (pgmq 1.7.0+)+popMessageEncoder :: E.Params PopMessage+popMessageEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #qty >$< E.param (E.nullable E.int4))++messageQueryEncoder :: E.Params MessageQuery+messageQueryEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageId >$< E.param (E.nonNullable messageIdValue))++batchMessageQueryEncoder :: E.Params BatchMessageQuery+batchMessageQueryEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageIds >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageIdValue))))))++visibilityTimeoutQueryEncoder :: E.Params VisibilityTimeoutQuery+visibilityTimeoutQueryEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageId >$< E.param (E.nonNullable messageIdValue))+    <> (view #visibilityTimeoutOffset >$< E.param (E.nonNullable E.int4))++-- | Encoder for BatchVisibilityTimeoutQuery (pgmq 1.8.0+)+batchVisibilityTimeoutQueryEncoder :: E.Params BatchVisibilityTimeoutQuery+batchVisibilityTimeoutQueryEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageIds >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageIdValue))))))+    <> (view #visibilityTimeoutOffset >$< E.param (E.nonNullable E.int4))++-- | Encoder for VisibilityTimeoutAtQuery (pgmq 1.10.0+)+-- SQL: pgmq.set_vt(queue_name, msg_id, timestamp)+visibilityTimeoutAtQueryEncoder :: E.Params VisibilityTimeoutAtQuery+visibilityTimeoutAtQueryEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageId >$< E.param (E.nonNullable messageIdValue))+    <> (view #visibilityTime >$< E.param (E.nonNullable E.timestamptz))++-- | Encoder for BatchVisibilityTimeoutAtQuery (pgmq 1.10.0+)+-- SQL: pgmq.set_vt(queue_name, msg_ids[], timestamp)+batchVisibilityTimeoutAtQueryEncoder :: E.Params BatchVisibilityTimeoutAtQuery+batchVisibilityTimeoutAtQueryEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #messageIds >$< E.param (E.nonNullable (E.array (E.dimension foldl' (E.element (E.nonNullable messageIdValue))))))+    <> (view #visibilityTime >$< E.param (E.nonNullable E.timestamptz))++-- | Encoder for EnableNotifyInsert (pgmq 1.7.0+)+enableNotifyInsertEncoder :: E.Params EnableNotifyInsert+enableNotifyInsertEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #throttleIntervalMs >$< E.param (E.nullable E.int4))++readWithPollEncoder :: E.Params ReadWithPollMessage+readWithPollEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #delay >$< E.param (E.nonNullable E.int4))+    <> (view #batchSize >$< E.param (E.nullable E.int4))+    <> (view #maxPollSeconds >$< E.param (E.nonNullable E.int4))+    <> (view #pollIntervalMs >$< E.param (E.nonNullable E.int4))+    <> (view #conditional >$< E.param (E.nullable E.jsonb))++createPartitionedQueueEncoder :: E.Params CreatePartitionedQueue+createPartitionedQueueEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #partitionInterval >$< E.param (E.nonNullable E.text))+    <> (view #retentionInterval >$< E.param (E.nonNullable E.text))++-- | Encoder for ReadGrouped (pgmq 1.8.0+)+-- Used for read_grouped and read_grouped_rr+readGroupedEncoder :: E.Params ReadGrouped+readGroupedEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #visibilityTimeout >$< E.param (E.nonNullable E.int4))+    <> (view #qty >$< E.param (E.nonNullable E.int4))++-- | Encoder for ReadGroupedWithPoll (pgmq 1.8.0+)+-- Used for read_grouped_with_poll and read_grouped_rr_with_poll+readGroupedWithPollEncoder :: E.Params ReadGroupedWithPoll+readGroupedWithPollEncoder =+  (view #queueName >$< E.param (E.nonNullable queueNameValue))+    <> (view #visibilityTimeout >$< E.param (E.nonNullable E.int4))+    <> (view #qty >$< E.param (E.nonNullable E.int4))+    <> (view #maxPollSeconds >$< E.param (E.nonNullable E.int4))+    <> (view #pollIntervalMs >$< E.param (E.nonNullable E.int4))

@@ -0,0 +1,63 @@+{-# LANGUAGE PackageImports #-}++module Pgmq.Hasql.Prelude+  ( -- base+    Generic,+    Data,+    Int32,+    Int64,+    void,+    when,+    unless,+    guard,+    fromMaybe,+    isJust,+    isNothing,+    CallStack,+    prettyCallStack,+    Proxy (..),+    (<|>),+    (>$<),+    MonadIO,+    NonEmpty,+    -- text+    Text,+    -- aeson+    FromJSON,+    ToJSON,+    parseJSON,+    genericParseJSON,+    genericToJSON,+    genericToEncoding,+    toJSON,+    fromJSON,+    toEncoding,+    -- time+    UTCTime,+    Day,+    LocalTime,+    getCurrentTime,+    -- lens+    module Control.Lens,+    -- vector+    Vector,+  )+where++import "aeson" Data.Aeson+import "base" Control.Applicative ((<|>))+import "base" Control.Monad (guard, unless, void, when)+import "base" Control.Monad.IO.Class (MonadIO)+import "base" Data.Data (Data)+import "base" Data.Functor.Contravariant ((>$<))+import "base" Data.Int (Int32, Int64)+import "base" Data.List.NonEmpty (NonEmpty)+import "base" Data.Maybe (fromMaybe, isJust, isNothing)+import "base" Data.Proxy (Proxy (..))+import "base" GHC.Exception (CallStack, prettyCallStack)+import "base" GHC.Generics (Generic)+import "generic-lens" Data.Generics.Labels ()+import "lens" Control.Lens+import "text" Data.Text (Text)+import "time" Data.Time (Day, LocalTime, UTCTime, getCurrentTime)+import "vector" Data.Vector (Vector)

@@ -0,0 +1,16 @@+{-# LANGUAGE TemplateHaskellQuotes #-}++module Pgmq.Hasql.Quasi (pgmq) where++import Data.Text qualified as T+import Language.Haskell.TH+import Language.Haskell.TH.Quote+import Pgmq.Types (parseQueueName)++pgmq :: QuasiQuoter+pgmq = QuasiQuoter {quoteExp = mkQueueName, quotePat = undefined, quoteType = undefined, quoteDec = undefined}++mkQueueName :: String -> Q Exp+mkQueueName str = case parseQueueName (T.pack str) of+  Right q -> [|q|]+  Left e -> fail $ show e

@@ -0,0 +1,198 @@+module Pgmq.Hasql.Sessions+  ( createQueue,+    dropQueue,+    createPartitionedQueue,+    createUnloggedQueue,+    detachArchive,+    enableNotifyInsert,+    disableNotifyInsert,+    sendMessage,+    sendMessageForLater,+    batchSendMessage,+    batchSendMessageForLater,+    sendMessageWithHeaders,+    sendMessageWithHeadersForLater,+    batchSendMessageWithHeaders,+    batchSendMessageWithHeadersForLater,+    deleteMessage,+    batchDeleteMessages,+    archiveMessage,+    batchArchiveMessages,+    deleteAllMessagesFromQueue,+    changeVisibilityTimeout,+    batchChangeVisibilityTimeout,+    -- Timestamp-based VT functions (pgmq 1.10.0+)+    setVisibilityTimeoutAt,+    batchSetVisibilityTimeoutAt,+    listQueues,+    pop,+    queueMetrics,+    allQueueMetrics,+    readMessage,+    readWithPoll,+    -- FIFO read functions (pgmq 1.8.0+)+    readGrouped,+    readGroupedWithPoll,+    -- Round-robin FIFO functions (pgmq 1.9.0+)+    readGroupedRoundRobin,+    readGroupedRoundRobinWithPoll,+    -- FIFO index functions (pgmq 1.8.0+)+    createFifoIndex,+    createFifoIndexesAll,+  )+where++import Hasql.Session (Session, statement)+import Pgmq.Hasql.Prelude+import Pgmq.Hasql.Statements qualified as Stmt+import Pgmq.Hasql.Statements.Message qualified as Msg+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)++createQueue :: QueueName -> Session ()+createQueue q = statement q Stmt.createQueue++dropQueue :: QueueName -> Session Bool+dropQueue q = statement q Stmt.dropQueue++sendMessage :: SendMessage -> Session MessageId+sendMessage msg = statement msg Msg.sendMessage++sendMessageForLater :: SendMessageForLater -> Session MessageId+sendMessageForLater msg = statement msg Msg.sendMessageForLater++batchSendMessage :: BatchSendMessage -> Session [MessageId]+batchSendMessage msgs = statement msgs Msg.batchSendMessage++batchSendMessageForLater :: BatchSendMessageForLater -> Session [MessageId]+batchSendMessageForLater msgs = statement msgs Msg.batchSendMessageForLater++-- | Send a message with headers (pgmq 1.5.0+)+sendMessageWithHeaders :: SendMessageWithHeaders -> Session MessageId+sendMessageWithHeaders msg = statement msg Msg.sendMessageWithHeaders++-- | Send a message with headers for later (pgmq 1.5.0+)+sendMessageWithHeadersForLater :: SendMessageWithHeadersForLater -> Session MessageId+sendMessageWithHeadersForLater msg = statement msg Msg.sendMessageWithHeadersForLater++-- | Send a batch of messages with headers (pgmq 1.5.0+)+batchSendMessageWithHeaders :: BatchSendMessageWithHeaders -> Session [MessageId]+batchSendMessageWithHeaders msgs = statement msgs Msg.batchSendMessageWithHeaders++-- | Send a batch of messages with headers for later (pgmq 1.5.0+)+batchSendMessageWithHeadersForLater :: BatchSendMessageWithHeadersForLater -> Session [MessageId]+batchSendMessageWithHeadersForLater msgs = statement msgs Msg.batchSendMessageWithHeadersForLater++deleteMessage :: MessageQuery -> Session Bool+deleteMessage msg = statement msg Msg.deleteMessage++batchDeleteMessages :: BatchMessageQuery -> Session [MessageId]+batchDeleteMessages msgs = statement msgs Msg.batchDeleteMessages++archiveMessage :: MessageQuery -> Session Bool+archiveMessage msg = statement msg Msg.archiveMessage++batchArchiveMessages :: BatchMessageQuery -> Session [MessageId]+batchArchiveMessages msgs = statement msgs Msg.batchArchiveMessages++deleteAllMessagesFromQueue :: QueueName -> Session Int64+deleteAllMessagesFromQueue qname = statement qname Msg.deleteAllMessagesFromQueue++changeVisibilityTimeout :: VisibilityTimeoutQuery -> Session Message+changeVisibilityTimeout query = statement query Msg.changeVisibilityTimeout++-- | Batch update visibility timeout (pgmq 1.8.0+)+batchChangeVisibilityTimeout :: BatchVisibilityTimeoutQuery -> Session (Vector Message)+batchChangeVisibilityTimeout query = statement query Msg.batchChangeVisibilityTimeout++-- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+)+setVisibilityTimeoutAt :: VisibilityTimeoutAtQuery -> Session Message+setVisibilityTimeoutAt query = statement query Msg.setVisibilityTimeoutAt++-- | Batch set visibility timeout to an absolute timestamp (pgmq 1.10.0+)+batchSetVisibilityTimeoutAt :: BatchVisibilityTimeoutAtQuery -> Session (Vector Message)+batchSetVisibilityTimeoutAt query = statement query Msg.batchSetVisibilityTimeoutAt++listQueues :: Session [Queue]+listQueues = statement () Stmt.listQueues++createPartitionedQueue :: CreatePartitionedQueue -> Session ()+createPartitionedQueue q = statement q Stmt.createPartitionedQueue++createUnloggedQueue :: QueueName -> Session ()+createUnloggedQueue q = statement q Stmt.createUnloggedQueue++{-# DEPRECATED detachArchive "detach_archive is a no-op in pgmq and will be removed in pgmq 2.0" #-}+detachArchive :: QueueName -> Session ()+detachArchive q = statement q Stmt.detachArchive++-- | Enable insert notifications for a queue (pgmq 1.7.0+)+enableNotifyInsert :: EnableNotifyInsert -> Session ()+enableNotifyInsert config = statement config Stmt.enableNotifyInsert++-- | Disable insert notifications for a queue+disableNotifyInsert :: QueueName -> Session ()+disableNotifyInsert q = statement q Stmt.disableNotifyInsert++-- | Pop messages from queue (pgmq 1.7.0+)+pop :: PopMessage -> Session (Vector Message)+pop query = statement query Msg.pop++queueMetrics :: QueueName -> Session QueueMetrics+queueMetrics q = statement q Stmt.queueMetrics++allQueueMetrics :: Session [QueueMetrics]+allQueueMetrics = statement () Stmt.allQueueMetrics++readMessage :: ReadMessage -> Session (Vector Message)+readMessage query = statement query Stmt.readMessage++readWithPoll :: ReadWithPollMessage -> Session (Vector Message)+readWithPoll query = statement query Stmt.readWithPoll++-- | FIFO read - fills batch from same message group (pgmq 1.8.0+)+readGrouped :: ReadGrouped -> Session (Vector Message)+readGrouped query = statement query Msg.readGrouped++-- | FIFO read with polling (pgmq 1.8.0+)+readGroupedWithPoll :: ReadGroupedWithPoll -> Session (Vector Message)+readGroupedWithPoll query = statement query Msg.readGroupedWithPoll++-- | Round-robin FIFO read (pgmq 1.9.0+)+readGroupedRoundRobin :: ReadGrouped -> Session (Vector Message)+readGroupedRoundRobin query = statement query Msg.readGroupedRoundRobin++-- | Round-robin FIFO read with polling (pgmq 1.9.0+)+readGroupedRoundRobinWithPoll :: ReadGroupedWithPoll -> Session (Vector Message)+readGroupedRoundRobinWithPoll query = statement query Msg.readGroupedRoundRobinWithPoll++-- | Create FIFO index for a queue (pgmq 1.8.0+)+createFifoIndex :: QueueName -> Session ()+createFifoIndex q = statement q Stmt.createFifoIndex++-- | Create FIFO indexes for all queues (pgmq 1.8.0+)+createFifoIndexesAll :: Session ()+createFifoIndexesAll = statement () Stmt.createFifoIndexesAll

@@ -0,0 +1,10 @@+module Pgmq.Hasql.Statements+  ( module Pgmq.Hasql.Statements.Message,+    module Pgmq.Hasql.Statements.QueueManagement,+    module Pgmq.Hasql.Statements.QueueObservability,+  )+where++import Pgmq.Hasql.Statements.Message+import Pgmq.Hasql.Statements.QueueManagement+import Pgmq.Hasql.Statements.QueueObservability

@@ -0,0 +1,272 @@+module Pgmq.Hasql.Statements.Message+  ( sendMessage,+    sendMessageForLater,+    batchSendMessage,+    batchSendMessageForLater,+    sendMessageWithHeaders,+    sendMessageWithHeadersForLater,+    batchSendMessageWithHeaders,+    batchSendMessageWithHeadersForLater,+    readMessage,+    deleteMessage,+    batchDeleteMessages,+    archiveMessage,+    batchArchiveMessages,+    deleteAllMessagesFromQueue,+    changeVisibilityTimeout,+    batchChangeVisibilityTimeout,+    -- Timestamp-based VT functions (pgmq 1.10.0+)+    setVisibilityTimeoutAt,+    batchSetVisibilityTimeoutAt,+    readWithPoll,+    pop,+    -- FIFO read functions (pgmq 1.8.0+)+    readGrouped,+    readGroupedWithPoll,+    -- Round-robin FIFO functions (pgmq 1.9.0+)+    readGroupedRoundRobin,+    readGroupedRoundRobinWithPoll,+  )+where++import Hasql.Decoders qualified as D+import Hasql.Statement (Statement, preparable)+import Pgmq.Hasql.Decoders (messageDecoder, messageIdDecoder)+import Pgmq.Hasql.Encoders+  ( batchMessageQueryEncoder,+    batchSendMessageEncoder,+    batchSendMessageForLaterEncoder,+    batchSendMessageWithHeadersEncoder,+    batchSendMessageWithHeadersForLaterEncoder,+    batchVisibilityTimeoutAtQueryEncoder,+    batchVisibilityTimeoutQueryEncoder,+    messageQueryEncoder,+    popMessageEncoder,+    queueNameEncoder,+    readGroupedEncoder,+    readGroupedWithPollEncoder,+    readMessageEncoder,+    readWithPollEncoder,+    sendMessageEncoder,+    sendMessageForLaterEncoder,+    sendMessageWithHeadersEncoder,+    sendMessageWithHeadersForLaterEncoder,+    visibilityTimeoutAtQueryEncoder,+    visibilityTimeoutQueryEncoder,+  )+import Pgmq.Hasql.Prelude+import Pgmq.Hasql.Statements.Types+  ( BatchMessageQuery,+    BatchSendMessage,+    BatchSendMessageForLater,+    BatchSendMessageWithHeaders,+    BatchSendMessageWithHeadersForLater,+    BatchVisibilityTimeoutAtQuery,+    BatchVisibilityTimeoutQuery,+    MessageQuery,+    PopMessage,+    ReadGrouped,+    ReadGroupedWithPoll,+    ReadMessage,+    ReadWithPollMessage,+    SendMessage,+    SendMessageForLater,+    SendMessageWithHeaders,+    SendMessageWithHeadersForLater,+    VisibilityTimeoutAtQuery,+    VisibilityTimeoutQuery,+  )+import Pgmq.Types (Message, MessageId, QueueName)++-- https://tembo.io/pgmq/api/sql/functions/#send+-- Note: coalesce handles null delay to ensure correct function overload resolution+sendMessage :: Statement SendMessage MessageId+sendMessage = preparable sql sendMessageEncoder decoder+  where+    sql = "select * from pgmq.send($1, $2, coalesce($3, 0))"+    decoder = D.singleRow messageIdDecoder++-- https://tembo.io/pgmq/api/sql/functions/#send+sendMessageForLater :: Statement SendMessageForLater MessageId+sendMessageForLater = preparable sql sendMessageForLaterEncoder decoder+  where+    sql = "select * from pgmq.send($1, $2, $3)"+    decoder = D.singleRow messageIdDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#send_batch+-- Note: coalesce handles null delay to ensure correct function overload resolution+batchSendMessage :: Statement BatchSendMessage [MessageId]+batchSendMessage = preparable sql batchSendMessageEncoder decoder+  where+    sql = "select * from pgmq.send_batch($1, $2, coalesce($3, 0))"+    decoder = D.rowList messageIdDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#send_batch+batchSendMessageForLater :: Statement BatchSendMessageForLater [MessageId]+batchSendMessageForLater = preparable sql batchSendMessageForLaterEncoder decoder+  where+    sql = "select * from pgmq.send_batch($1, $2, $3)"+    decoder = D.rowList messageIdDecoder++-- | Send a message with headers (pgmq 1.5.0+)+-- https://tembo.io/pgmq/api/sql/functions/#send+-- Note: coalesce handles null delay to ensure correct function overload resolution+sendMessageWithHeaders :: Statement SendMessageWithHeaders MessageId+sendMessageWithHeaders = preparable sql sendMessageWithHeadersEncoder decoder+  where+    sql = "select * from pgmq.send($1, $2, $3, coalesce($4, 0))"+    decoder = D.singleRow messageIdDecoder++-- | Send a message with headers for later (pgmq 1.5.0+)+-- https://tembo.io/pgmq/api/sql/functions/#send+sendMessageWithHeadersForLater :: Statement SendMessageWithHeadersForLater MessageId+sendMessageWithHeadersForLater = preparable sql sendMessageWithHeadersForLaterEncoder decoder+  where+    sql = "select * from pgmq.send($1, $2, $3, $4)"+    decoder = D.singleRow messageIdDecoder++-- | Send a batch of messages with headers (pgmq 1.5.0+)+-- https://tembo.io/pgmq/api/sql/functions/#send_batch+-- Note: coalesce handles null delay to ensure correct function overload resolution+batchSendMessageWithHeaders :: Statement BatchSendMessageWithHeaders [MessageId]+batchSendMessageWithHeaders = preparable sql batchSendMessageWithHeadersEncoder decoder+  where+    sql = "select * from pgmq.send_batch($1, $2, $3, coalesce($4, 0))"+    decoder = D.rowList messageIdDecoder++-- | Send a batch of messages with headers for later (pgmq 1.5.0+)+-- https://tembo.io/pgmq/api/sql/functions/#send_batch+batchSendMessageWithHeadersForLater :: Statement BatchSendMessageWithHeadersForLater [MessageId]+batchSendMessageWithHeadersForLater = preparable sql batchSendMessageWithHeadersForLaterEncoder decoder+  where+    sql = "select * from pgmq.send_batch($1, $2, $3, $4)"+    decoder = D.rowList messageIdDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#read+-- Note: conditional parameter added in pgmq 1.5.0+-- We use the 3-param version since the 4-param version fails with NULL conditional+-- (message @> NULL = NULL, not TRUE, so no rows match).+-- To use conditional filtering, use readMessageConditional instead.+readMessage :: Statement ReadMessage (Vector Message)+readMessage = preparable sql readMessageEncoder decoder+  where+    sql = "select * from pgmq.read($1,$2,$3)"+    decoder = D.rowVector messageDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#delete-single+deleteMessage :: Statement MessageQuery Bool+deleteMessage = preparable sql messageQueryEncoder decoder+  where+    sql = "select * from pgmq.delete($1,$2)"+    decoder = D.singleRow (D.column (D.nonNullable D.bool))++-- | https://tembo.io/pgmq/api/sql/functions/#delete-batch+batchDeleteMessages :: Statement BatchMessageQuery [MessageId]+batchDeleteMessages = preparable sql batchMessageQueryEncoder decoder+  where+    sql = "select * from pgmq.delete($1,$2)"+    decoder = D.rowList messageIdDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#archive-single+archiveMessage :: Statement MessageQuery Bool+archiveMessage = preparable sql messageQueryEncoder decoder+  where+    sql = "select * from pgmq.archive($1,$2)"+    decoder = D.singleRow (D.column (D.nonNullable D.bool))++-- | https://tembo.io/pgmq/api/sql/functions/#archive-batch+batchArchiveMessages :: Statement BatchMessageQuery [MessageId]+batchArchiveMessages = preparable sql batchMessageQueryEncoder decoder+  where+    sql = "select * from pgmq.archive($1,$2)"+    decoder = D.rowList messageIdDecoder++-- | Permanently deletes all messages in a queue. Returns the number of messages that were deleted.+-- | https://tembo.io/pgmq/api/sql/functions/#purge_queue+deleteAllMessagesFromQueue :: Statement QueueName Int64+deleteAllMessagesFromQueue = preparable sql queueNameEncoder decoder+  where+    sql = "select * from pgmq.purge_queue($1)"+    decoder = D.singleRow $ D.column $ D.nonNullable D.int8++-- | Sets the visibility timeout of a message to a specified time duration in the future. Returns the record of the message that was updated.+-- | https://tembo.io/pgmq/api/sql/functions/#set_vt+changeVisibilityTimeout :: Statement VisibilityTimeoutQuery Message+changeVisibilityTimeout = preparable sql visibilityTimeoutQueryEncoder decoder+  where+    sql = "select * from pgmq.set_vt($1,$2,$3)"+    decoder = D.singleRow messageDecoder++-- | Batch update visibility timeout for multiple messages (pgmq 1.8.0+)+-- | https://tembo.io/pgmq/api/sql/functions/#set_vt+batchChangeVisibilityTimeout :: Statement BatchVisibilityTimeoutQuery (Vector Message)+batchChangeVisibilityTimeout = preparable sql batchVisibilityTimeoutQueryEncoder decoder+  where+    sql = "select * from pgmq.set_vt($1,$2,$3)"+    decoder = D.rowVector messageDecoder++-- | Set visibility timeout to an absolute timestamp (pgmq 1.10.0+)+-- | https://tembo.io/pgmq/api/sql/functions/#set_vt+setVisibilityTimeoutAt :: Statement VisibilityTimeoutAtQuery Message+setVisibilityTimeoutAt = preparable sql visibilityTimeoutAtQueryEncoder decoder+  where+    sql = "select * from pgmq.set_vt($1,$2,$3)"+    decoder = D.singleRow messageDecoder++-- | Batch set visibility timeout to an absolute timestamp (pgmq 1.10.0+)+-- | https://tembo.io/pgmq/api/sql/functions/#set_vt+batchSetVisibilityTimeoutAt :: Statement BatchVisibilityTimeoutAtQuery (Vector Message)+batchSetVisibilityTimeoutAt = preparable sql batchVisibilityTimeoutAtQueryEncoder decoder+  where+    sql = "select * from pgmq.set_vt($1,$2,$3)"+    decoder = D.rowVector messageDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#read_with_poll+readWithPoll :: Statement ReadWithPollMessage (Vector Message)+readWithPoll = preparable sql readWithPollEncoder decoder+  where+    sql = "select * from pgmq.read_with_poll($1,$2,$3,$4,$5,$6)"+    decoder = D.rowVector messageDecoder++-- | Pop messages from queue (atomic read + delete)+-- https://tembo.io/pgmq/api/sql/functions/#pop+-- Note: qty parameter added in pgmq 1.7.0+pop :: Statement PopMessage (Vector Message)+pop = preparable sql popMessageEncoder decoder+  where+    sql = "select * from pgmq.pop($1,$2)"+    decoder = D.rowVector messageDecoder++-- | FIFO read - fills batch from same message group (pgmq 1.8.0+)+-- Messages are grouped by the x-pgmq-group header.+-- https://tembo.io/pgmq/api/sql/functions/#read_grouped+readGrouped :: Statement ReadGrouped (Vector Message)+readGrouped = preparable sql readGroupedEncoder decoder+  where+    sql = "select * from pgmq.read_grouped($1,$2,$3)"+    decoder = D.rowVector messageDecoder++-- | FIFO read with polling - fills batch from same message group (pgmq 1.8.0+)+-- https://tembo.io/pgmq/api/sql/functions/#read_grouped_with_poll+readGroupedWithPoll :: Statement ReadGroupedWithPoll (Vector Message)+readGroupedWithPoll = preparable sql readGroupedWithPollEncoder decoder+  where+    sql = "select * from pgmq.read_grouped_with_poll($1,$2,$3,$4,$5)"+    decoder = D.rowVector messageDecoder++-- | Round-robin FIFO read - fair distribution across message groups (pgmq 1.9.0+)+-- Uses layered round-robin algorithm for fairness.+-- https://tembo.io/pgmq/api/sql/functions/#read_grouped_rr+readGroupedRoundRobin :: Statement ReadGrouped (Vector Message)+readGroupedRoundRobin = preparable sql readGroupedEncoder decoder+  where+    sql = "select * from pgmq.read_grouped_rr($1,$2,$3)"+    decoder = D.rowVector messageDecoder++-- | Round-robin FIFO read with polling (pgmq 1.9.0+)+-- https://tembo.io/pgmq/api/sql/functions/#read_grouped_rr_with_poll+readGroupedRoundRobinWithPoll :: Statement ReadGroupedWithPoll (Vector Message)+readGroupedRoundRobinWithPoll = preparable sql readGroupedWithPollEncoder decoder+  where+    sql = "select * from pgmq.read_grouped_rr_with_poll($1,$2,$3,$4,$5)"+    decoder = D.rowVector messageDecoder

@@ -0,0 +1,78 @@+module Pgmq.Hasql.Statements.QueueManagement+  ( createQueue,+    dropQueue,+    createPartitionedQueue,+    createUnloggedQueue,+    detachArchive,+    enableNotifyInsert,+    disableNotifyInsert,+    -- FIFO index functions (pgmq 1.8.0+)+    createFifoIndex,+    createFifoIndexesAll,+  )+where++import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (Statement, preparable)+import Pgmq.Hasql.Encoders (createPartitionedQueueEncoder, enableNotifyInsertEncoder, queueNameEncoder)+import Pgmq.Hasql.Statements.Types (CreatePartitionedQueue, EnableNotifyInsert)+import Pgmq.Types (QueueName)++-- https://tembo.io/pgmq/api/sql/functions/#create+createQueue :: Statement QueueName ()+createQueue = preparable sql queueNameEncoder D.noResult+  where+    sql = "select from pgmq.create($1)"++-- https://tembo.io/pgmq/api/sql/functions/#drop_queue+dropQueue :: Statement QueueName Bool+dropQueue = preparable sql queueNameEncoder decoder+  where+    sql = "select * from pgmq.drop_queue($1)"+    decoder = D.singleRow (D.column (D.nonNullable D.bool))++-- | https://tembo.io/pgmq/api/sql/functions/#create_partitioned+createPartitionedQueue :: Statement CreatePartitionedQueue ()+createPartitionedQueue = preparable sql createPartitionedQueueEncoder D.noResult+  where+    sql = "select from pgmq.create_partitioned($1,$2,$3)"++-- | https://tembo.io/pgmq/api/sql/functions/#create_unlogged+createUnloggedQueue :: Statement QueueName ()+createUnloggedQueue = preparable sql queueNameEncoder D.noResult+  where+    sql = "select from pgmq.create_unlogged($1)"++-- | DEPRECATED: detach_archive is a no-op in pgmq and will be removed in pgmq 2.0+-- https://tembo.io/pgmq/api/sql/functions/#detach_archive+detachArchive :: Statement QueueName ()+detachArchive = preparable sql queueNameEncoder D.noResult+  where+    sql = "select from pgmq.detach_archive($1)"++-- | Enable insert notifications for a queue (pgmq 1.7.0+)+-- Notifications are sent via PostgreSQL LISTEN/NOTIFY to channel pgmq_<queue_name>+enableNotifyInsert :: Statement EnableNotifyInsert ()+enableNotifyInsert = preparable sql enableNotifyInsertEncoder D.noResult+  where+    sql = "select from pgmq.enable_notify_insert($1, $2)"++-- | Disable insert notifications for a queue+disableNotifyInsert :: Statement QueueName ()+disableNotifyInsert = preparable sql queueNameEncoder D.noResult+  where+    sql = "select from pgmq.disable_notify_insert($1)"++-- | Create FIFO index for a queue (pgmq 1.8.0+)+-- Improves performance for FIFO read operations.+createFifoIndex :: Statement QueueName ()+createFifoIndex = preparable sql queueNameEncoder D.noResult+  where+    sql = "select from pgmq.create_fifo_index($1)"++-- | Create FIFO indexes for all queues (pgmq 1.8.0+)+createFifoIndexesAll :: Statement () ()+createFifoIndexesAll = preparable sql E.noParams D.noResult+  where+    sql = "select from pgmq.create_fifo_indexes_all()"

@@ -0,0 +1,36 @@+module Pgmq.Hasql.Statements.QueueObservability+  ( listQueues,+    queueMetrics,+    allQueueMetrics,+  )+where++import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (Statement, preparable)+import Pgmq.Hasql.Decoders (queueDecoder, queueMetricsDecoder)+import Pgmq.Hasql.Encoders (queueNameEncoder)+import Pgmq.Hasql.Statements.Types (QueueMetrics)+import Pgmq.Types (Queue, QueueName)++-- | List all queues that currently exist+-- | https://tembo.io/pgmq/api/sql/functions/#list_queues+listQueues :: Statement () [Queue]+listQueues = preparable sql E.noParams decoder+  where+    sql = "select * from pgmq.list_queues()"+    decoder = D.rowList queueDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#metrics+queueMetrics :: Statement QueueName QueueMetrics+queueMetrics = preparable sql queueNameEncoder decoder+  where+    sql = "select * from pgmq.metrics($1)"+    decoder = D.singleRow queueMetricsDecoder++-- | https://tembo.io/pgmq/api/sql/functions/#metrics_all+allQueueMetrics :: Statement () [QueueMetrics]+allQueueMetrics = preparable sql E.noParams decoder+  where+    sql = "select * from pgmq.metrics_all()"+    decoder = D.rowList queueMetricsDecoder

@@ -0,0 +1,220 @@+module Pgmq.Hasql.Statements.Types+  ( SendMessage (..),+    SendMessageForLater (..),+    BatchSendMessage (..),+    BatchSendMessageForLater (..),+    SendMessageWithHeaders (..),+    SendMessageWithHeadersForLater (..),+    BatchSendMessageWithHeaders (..),+    BatchSendMessageWithHeadersForLater (..),+    ReadMessage (..),+    PopMessage (..),+    MessageQuery (..),+    BatchMessageQuery (..),+    VisibilityTimeoutQuery (..),+    BatchVisibilityTimeoutQuery (..),+    -- Timestamp-based VT types (pgmq 1.10.0+)+    VisibilityTimeoutAtQuery (..),+    BatchVisibilityTimeoutAtQuery (..),+    ReadWithPollMessage (..),+    EnableNotifyInsert (..),+    CreatePartitionedQueue (..),+    QueueMetrics (..),+    -- FIFO read types (pgmq 1.8.0+)+    ReadGrouped (..),+    ReadGroupedWithPoll (..),+  )+where++import Data.Aeson (Value)+import Pgmq.Hasql.Prelude+import Pgmq.Types (MessageBody, MessageHeaders, MessageId, QueueName)++type Delay = Int32++data SendMessage = SendMessage+  { queueName :: !QueueName,+    messageBody :: !MessageBody,+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++data VisibilityTimeoutQuery = VisibilityTimeoutQuery+  { queueName :: !QueueName,+    messageId :: !MessageId,+    visibilityTimeoutOffset :: !Int32+  }+  deriving stock (Generic)++-- | Batch visibility timeout update (pgmq 1.8.0+)+data BatchVisibilityTimeoutQuery = BatchVisibilityTimeoutQuery+  { queueName :: !QueueName,+    messageIds :: ![MessageId],+    visibilityTimeoutOffset :: !Int32+  }+  deriving stock (Generic)++-- | Set visibility timeout to absolute timestamp (pgmq 1.10.0+)+data VisibilityTimeoutAtQuery = VisibilityTimeoutAtQuery+  { queueName :: !QueueName,+    messageId :: !MessageId,+    visibilityTime :: !UTCTime+  }+  deriving stock (Generic)++-- | Batch set visibility timeout to absolute timestamp (pgmq 1.10.0+)+data BatchVisibilityTimeoutAtQuery = BatchVisibilityTimeoutAtQuery+  { queueName :: !QueueName,+    messageIds :: ![MessageId],+    visibilityTime :: !UTCTime+  }+  deriving stock (Generic)++data BatchMessageQuery = BatchMessageQuery+  { queueName :: !QueueName,+    messageIds :: ![MessageId]+  }+  deriving stock (Generic)++data MessageQuery = MessageQuery+  { queueName :: !QueueName,+    messageId :: !MessageId+  }+  deriving stock (Generic)++data SendMessageForLater = SendMessageForLater+  { queueName :: !QueueName,+    messageBody :: !MessageBody,+    scheduledAt :: !UTCTime+  }+  deriving stock (Generic)++data BatchSendMessage = BatchSendMessage+  { queueName :: !QueueName,+    messageBodies :: ![MessageBody],+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++data BatchSendMessageForLater = BatchSendMessageForLater+  { queueName :: !QueueName,+    messageBodies :: ![MessageBody],+    scheduledAt :: !UTCTime+  }+  deriving stock (Generic)++-- | Send message with headers (pgmq 1.5.0+)+data SendMessageWithHeaders = SendMessageWithHeaders+  { queueName :: !QueueName,+    messageBody :: !MessageBody,+    messageHeaders :: !MessageHeaders,+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++-- | Send message with headers for later (pgmq 1.5.0+)+data SendMessageWithHeadersForLater = SendMessageWithHeadersForLater+  { queueName :: !QueueName,+    messageBody :: !MessageBody,+    messageHeaders :: !MessageHeaders,+    scheduledAt :: !UTCTime+  }+  deriving stock (Generic)++-- | Batch send messages with headers (pgmq 1.5.0+)+data BatchSendMessageWithHeaders = BatchSendMessageWithHeaders+  { queueName :: !QueueName,+    messageBodies :: ![MessageBody],+    messageHeaders :: ![MessageHeaders],+    delay :: !(Maybe Delay)+  }+  deriving stock (Generic)++-- | Batch send messages with headers for later (pgmq 1.5.0+)+data BatchSendMessageWithHeadersForLater = BatchSendMessageWithHeadersForLater+  { queueName :: !QueueName,+    messageBodies :: ![MessageBody],+    messageHeaders :: ![MessageHeaders],+    scheduledAt :: !UTCTime+  }+  deriving stock (Generic)++-- | Parameters for reading messages from a queue+-- Note: conditional field added in pgmq 1.5.0+data ReadMessage = ReadMessage+  { queueName :: !QueueName,+    delay :: !Delay,+    batchSize :: !(Maybe Int32),+    -- | Optional JSONB filter (pgmq 1.5.0+)+    conditional :: !(Maybe Value)+  }+  deriving stock (Generic)++data ReadWithPollMessage = ReadWithPollMessage+  { queueName :: !QueueName,+    delay :: !Delay,+    batchSize :: !(Maybe Int32),+    maxPollSeconds :: !Int32,+    pollIntervalMs :: !Int32,+    conditional :: !(Maybe Value)+  }+  deriving stock (Generic)++-- | Parameters for popping messages from a queue (pgmq 1.7.0+)+data PopMessage = PopMessage+  { queueName :: !QueueName,+    -- | Number of messages to pop (Nothing = default 1)+    qty :: !(Maybe Int32)+  }+  deriving stock (Generic)++-- | Enable queue notifications (pgmq 1.7.0+, throttling in 1.8.0+)+data EnableNotifyInsert = EnableNotifyInsert+  { queueName :: !QueueName,+    -- | Minimum ms between notifications (Nothing = default 250ms)+    throttleIntervalMs :: !(Maybe Int32)+  }+  deriving stock (Generic)++data CreatePartitionedQueue = CreatePartitionedQueue+  { queueName :: !QueueName,+    partitionInterval :: !Text,+    retentionInterval :: !Text+  }+  deriving stock (Generic)++-- | Queue metrics returned by pgmq.metrics() and pgmq.metrics_all()+-- Note: queueVisibleLength added in pgmq 1.5.0+data QueueMetrics = QueueMetrics+  { queueName :: !Text,+    queueLength :: !Int64,+    newestMsgAgeSec :: !(Maybe Int32),+    oldestMsgAgeSec :: !(Maybe Int32),+    totalMessages :: !Int64,+    scrapeTime :: !UTCTime,+    -- | Count of messages available for reading (pgmq 1.5.0+)+    queueVisibleLength :: !Int64+  }+  deriving stock (Generic, Show)++-- | Parameters for FIFO grouped read (pgmq 1.8.0+)+-- Used for both read_grouped and read_grouped_rr functions.+-- Note: conditional parameter was removed in pgmq 1.9.0 (commit 9e9c3dc)+data ReadGrouped = ReadGrouped+  { queueName :: !QueueName,+    visibilityTimeout :: !Int32,+    qty :: !Int32+  }+  deriving stock (Generic)++-- | Parameters for FIFO grouped read with polling (pgmq 1.8.0+)+-- Used for both read_grouped_with_poll and read_grouped_rr_with_poll functions.+-- Note: conditional parameter was removed in pgmq 1.9.0 (commit 9e9c3dc)+data ReadGroupedWithPoll = ReadGroupedWithPoll+  { queueName :: !QueueName,+    visibilityTimeout :: !Int32,+    qty :: !Int32,+    maxPollSeconds :: !Int32,+    pollIntervalMs :: !Int32+  }+  deriving stock (Generic)

@@ -0,0 +1,378 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests for advanced message operations:+-- - pop with qty+-- - batch set_vt+-- - read_with_poll+-- - FIFO index functions+-- - read_grouped functions+module AdvancedOpsSpec (tests) where++import Control.Concurrent (threadDelay)+import Data.Aeson (object, (.=))+import Data.Time.Clock (addUTCTime, getCurrentTime)+import Data.Vector qualified as V+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( BatchMessageQuery (..),+    BatchSendMessage (..),+    BatchSendMessageWithHeaders (..),+    BatchVisibilityTimeoutAtQuery (..),+    BatchVisibilityTimeoutQuery (..),+    PopMessage (..),+    ReadGrouped (..),+    ReadGroupedWithPoll (..),+    ReadMessage (..),+    ReadWithPollMessage (..),+    SendMessage (..),+    VisibilityTimeoutAtQuery (..),+  )+import Pgmq.Types (MessageBody (..), MessageHeaders (..), MessageId (..))+import Pgmq.Types qualified as PgmqTypes+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, testCase)+import TestUtils (assertSession, cleanupQueue)++-- | All advanced operation tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Advanced Operations"+    [ testPopSingle p,+      testPopBatch p,+      testBatchChangeVisibilityTimeout p,+      testSetVisibilityTimeoutAt p,+      testBatchSetVisibilityTimeoutAt p,+      testReadWithPoll p,+      testReadWithPollEmpty p,+      testCreateFifoIndex p,+      testReadGrouped p,+      testReadGroupedRoundRobin p+    ]++-- | Test pop with default qty (single message)+testPopSingle :: Pool.Pool -> TestTree+testPopSingle p = testCase "pop returns and deletes single message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send messages+    let msgs =+          BatchSendMessage+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["pop" .= (1 :: Int)]),+                  MessageBody (object ["pop" .= (2 :: Int)])+                ],+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.batchSendMessage msgs)+    -- Pop single message+    let popQuery = PopMessage {queueName = queueName, qty = Just 1}+    popped <- assertSession pool (Sessions.pop popQuery)+    assertEqual "Should pop 1 message" 1 (V.length popped)+    -- Verify only 1 message remains+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              conditional = Nothing+            }+    remaining <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should have 1 message remaining" 1 (V.length remaining)+    cleanupQueue pool queueName++-- | Test pop with qty > 1 (batch pop)+testPopBatch :: Pool.Pool -> TestTree+testPopBatch p = testCase "pop with qty > 1 returns multiple messages" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send 5 messages+    let msgs =+          BatchSendMessage+            { queueName = queueName,+              messageBodies = [MessageBody (object ["pop" .= i]) | i <- [1 .. 5 :: Int]],+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.batchSendMessage msgs)+    -- Pop 3 messages+    let popQuery = PopMessage {queueName = queueName, qty = Just 3}+    popped <- assertSession pool (Sessions.pop popQuery)+    assertEqual "Should pop 3 messages" 3 (V.length popped)+    -- Verify 2 messages remain+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              conditional = Nothing+            }+    remaining <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should have 2 messages remaining" 2 (V.length remaining)+    cleanupQueue pool queueName++-- | Test batchChangeVisibilityTimeout+testBatchChangeVisibilityTimeout :: Pool.Pool -> TestTree+testBatchChangeVisibilityTimeout p = testCase "batchChangeVisibilityTimeout updates multiple messages" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send messages+    let msgs =+          BatchSendMessage+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["vt" .= (1 :: Int)]),+                  MessageBody (object ["vt" .= (2 :: Int)]),+                  MessageBody (object ["vt" .= (3 :: Int)])+                ],+              delay = Nothing+            }+    msgIds <- assertSession pool (Sessions.batchSendMessage msgs)+    -- Read messages to set VT+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 5,+              batchSize = Just 10,+              conditional = Nothing+            }+    _ <- assertSession pool (Sessions.readMessage readQuery)+    -- Change VT for all messages+    let vtQuery =+          BatchVisibilityTimeoutQuery+            { queueName = queueName,+              messageIds = msgIds,+              visibilityTimeoutOffset = 60+            }+    updated <- assertSession pool (Sessions.batchChangeVisibilityTimeout vtQuery)+    assertEqual "Should update 3 messages" 3 (V.length updated)+    cleanupQueue pool queueName++-- | Test setVisibilityTimeoutAt (pgmq 1.10.0+)+testSetVisibilityTimeoutAt :: Pool.Pool -> TestTree+testSetVisibilityTimeoutAt p = testCase "setVisibilityTimeoutAt sets VT to absolute timestamp" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send a message+    let msg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["vt_at" .= ("test" :: String)]),+              delay = Nothing+            }+    msgId <- assertSession pool (Sessions.sendMessage msg)+    -- Read message to set initial VT+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 5,+              batchSize = Just 1,+              conditional = Nothing+            }+    _ <- assertSession pool (Sessions.readMessage readQuery)+    -- Set VT to 60 seconds in the future using absolute timestamp+    futureTime <- addUTCTime 60 <$> getCurrentTime+    let vtQuery =+          VisibilityTimeoutAtQuery+            { queueName = queueName,+              messageId = msgId,+              visibilityTime = futureTime+            }+    updated <- assertSession pool (Sessions.setVisibilityTimeoutAt vtQuery)+    assertEqual "Should return the updated message" msgId (PgmqTypes.messageId updated)+    cleanupQueue pool queueName++-- | Test batchSetVisibilityTimeoutAt (pgmq 1.10.0+)+testBatchSetVisibilityTimeoutAt :: Pool.Pool -> TestTree+testBatchSetVisibilityTimeoutAt p = testCase "batchSetVisibilityTimeoutAt sets VT for multiple messages" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send messages+    let msgs =+          BatchSendMessage+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["vt_at" .= (1 :: Int)]),+                  MessageBody (object ["vt_at" .= (2 :: Int)]),+                  MessageBody (object ["vt_at" .= (3 :: Int)])+                ],+              delay = Nothing+            }+    msgIds <- assertSession pool (Sessions.batchSendMessage msgs)+    -- Read messages to set initial VT+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 5,+              batchSize = Just 10,+              conditional = Nothing+            }+    _ <- assertSession pool (Sessions.readMessage readQuery)+    -- Set VT to 120 seconds in the future using absolute timestamp+    futureTime <- addUTCTime 120 <$> getCurrentTime+    let vtQuery =+          BatchVisibilityTimeoutAtQuery+            { queueName = queueName,+              messageIds = msgIds,+              visibilityTime = futureTime+            }+    updated <- assertSession pool (Sessions.batchSetVisibilityTimeoutAt vtQuery)+    assertEqual "Should update 3 messages" 3 (V.length updated)+    cleanupQueue pool queueName++-- | Test readWithPoll - messages available immediately+testReadWithPoll :: Pool.Pool -> TestTree+testReadWithPoll p = testCase "readWithPoll returns messages immediately when available" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send a message+    let msg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["poll" .= ("test" :: String)]),+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.sendMessage msg)+    -- Poll for messages+    let pollQuery =+          ReadWithPollMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              maxPollSeconds = 5,+              pollIntervalMs = 100,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readWithPoll pollQuery)+    assertEqual "Should read 1 message" 1 (V.length messages)+    cleanupQueue pool queueName++-- | Test readWithPoll - empty queue with short timeout+testReadWithPollEmpty :: Pool.Pool -> TestTree+testReadWithPollEmpty p = testCase "readWithPoll returns empty when queue is empty" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Poll on empty queue with short timeout+    let pollQuery =+          ReadWithPollMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              maxPollSeconds = 1, -- Short timeout+              pollIntervalMs = 100,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readWithPoll pollQuery)+    assertEqual "Should return empty" 0 (V.length messages)+    cleanupQueue pool queueName++-- | Test createFifoIndex+testCreateFifoIndex :: Pool.Pool -> TestTree+testCreateFifoIndex p = testCase "createFifoIndex creates GIN index on headers" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Create FIFO index+    assertSession pool (Sessions.createFifoIndex queueName)+    -- Verify by sending a message with headers (index should be used)+    let msg =+          BatchSendMessageWithHeaders+            { queueName = queueName,+              messageBodies = [MessageBody (object ["fifo" .= ("test" :: String)])],+              messageHeaders = [MessageHeaders (object ["x-pgmq-group" .= ("group1" :: String)])],+              delay = Nothing+            }+    msgIds <- assertSession pool (Sessions.batchSendMessageWithHeaders msg)+    assertEqual "Should send 1 message" 1 (length msgIds)+    cleanupQueue pool queueName++-- | Test readGrouped - fills batch from same message group+testReadGrouped :: Pool.Pool -> TestTree+testReadGrouped p = testCase "readGrouped fills batch from same message group" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Create FIFO index for better performance+    assertSession pool (Sessions.createFifoIndex queueName)+    -- Send messages from multiple groups+    let msgs =+          BatchSendMessageWithHeaders+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["msg" .= (1 :: Int)]),+                  MessageBody (object ["msg" .= (2 :: Int)]),+                  MessageBody (object ["msg" .= (3 :: Int)]),+                  MessageBody (object ["msg" .= (4 :: Int)])+                ],+              messageHeaders =+                [ MessageHeaders (object ["x-pgmq-group" .= ("groupA" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupA" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupB" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupB" :: String)])+                ],+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.batchSendMessageWithHeaders msgs)+    -- Read grouped - should prefer filling from one group+    let readQuery =+          ReadGrouped+            { queueName = queueName,+              visibilityTimeout = 30,+              qty = 2+            }+    messages <- assertSession pool (Sessions.readGrouped readQuery)+    assertEqual "Should read 2 messages" 2 (V.length messages)+    -- Delete the messages+    let msgIds = V.toList $ V.map PgmqTypes.messageId messages+    _ <-+      assertSession pool $+        Sessions.batchDeleteMessages+          BatchMessageQuery {queueName = queueName, messageIds = msgIds}+    cleanupQueue pool queueName++-- | Test readGroupedRoundRobin - interleaves across groups+testReadGroupedRoundRobin :: Pool.Pool -> TestTree+testReadGroupedRoundRobin p = testCase "readGroupedRoundRobin interleaves across groups" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Create FIFO index+    assertSession pool (Sessions.createFifoIndex queueName)+    -- Send messages from 3 groups+    let msgs =+          BatchSendMessageWithHeaders+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["msg" .= (1 :: Int)]),+                  MessageBody (object ["msg" .= (2 :: Int)]),+                  MessageBody (object ["msg" .= (3 :: Int)]),+                  MessageBody (object ["msg" .= (4 :: Int)]),+                  MessageBody (object ["msg" .= (5 :: Int)]),+                  MessageBody (object ["msg" .= (6 :: Int)])+                ],+              messageHeaders =+                [ MessageHeaders (object ["x-pgmq-group" .= ("groupA" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupB" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupC" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupA" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupB" :: String)]),+                  MessageHeaders (object ["x-pgmq-group" .= ("groupC" :: String)])+                ],+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.batchSendMessageWithHeaders msgs)+    -- Read round-robin - should interleave groups+    let readQuery =+          ReadGrouped+            { queueName = queueName,+              visibilityTimeout = 30,+              qty = 3+            }+    messages <- assertSession pool (Sessions.readGroupedRoundRobin readQuery)+    assertEqual "Should read 3 messages" 3 (V.length messages)+    -- Clean up+    let msgIds = V.toList $ V.map PgmqTypes.messageId messages+    _ <-+      assertSession pool $+        Sessions.batchDeleteMessages+          BatchMessageQuery {queueName = queueName, messageIds = msgIds}+    cleanupQueue pool queueName

@@ -0,0 +1,202 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Decoder tests for all pgmq functions that return messages+-- Each test verifies that the decoder works correctly for a specific function+module AllFunctionsDecoderSpec (tests) where++import Data.Aeson (Value, object, (.=))+import Data.Time (addUTCTime, getCurrentTime)+import Data.Vector qualified as V+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( BatchVisibilityTimeoutQuery (..),+    PopMessage (..),+    ReadMessage (..),+    ReadWithPollMessage (..),+    SendMessage (..),+    VisibilityTimeoutQuery (..),+  )+import Pgmq.Types (Message (..), MessageBody (..), QueueName, unMessageId)+import Pgmq.Types qualified+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, testCase)+import TestUtils (assertSession, cleanupQueue)++-- | All per-function decoder tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "All Functions Decoder Tests"+    [ testReadDecoder p,+      testPopDecoder p,+      testSetVtDecoder p,+      testBatchSetVtDecoder p,+      testReadWithPollDecoder p+    ]++-- | Test pgmq.read() decoder+testReadDecoder :: Pool.Pool -> TestTree+testReadDecoder p = testCase "read() returns correctly decoded message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message with known body+    let sentBody = object ["function" .= ("read" :: String), "value" .= (123 :: Int)]+    _ <- assertSession pool (Sessions.sendMessage (mkSendMessage queueName sentBody))++    -- Read via pgmq.read+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)++    assertEqual "Should return 1 message" 1 (V.length messages)+    let msg = V.head messages+    assertValidMessage msg sentBody++    cleanupQueue pool queueName++-- | Test pgmq.pop() decoder+testPopDecoder :: Pool.Pool -> TestTree+testPopDecoder p = testCase "pop() returns correctly decoded message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sentBody = object ["function" .= ("pop" :: String), "value" .= (456 :: Int)]+    _ <- assertSession pool (Sessions.sendMessage (mkSendMessage queueName sentBody))++    -- Pop message+    let popQuery = PopMessage {queueName = queueName, qty = Just 1}+    messages <- assertSession pool (Sessions.pop popQuery)++    assertEqual "Should return 1 message" 1 (V.length messages)+    let msg = V.head messages+    assertValidMessage msg sentBody++    -- Verify message is deleted (pop removes from queue)+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messagesAfter <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Queue should be empty after pop" 0 (V.length messagesAfter)++    cleanupQueue pool queueName++-- | Test pgmq.set_vt() decoder+testSetVtDecoder :: Pool.Pool -> TestTree+testSetVtDecoder p = testCase "set_vt() returns correctly decoded message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sentBody = object ["function" .= ("set_vt" :: String)]+    msgId <- assertSession pool (Sessions.sendMessage (mkSendMessage queueName sentBody))++    beforeSetVt <- getCurrentTime++    -- Change visibility timeout+    let vtQuery =+          VisibilityTimeoutQuery+            { queueName = queueName,+              messageId = msgId,+              visibilityTimeoutOffset = 60+            }+    msg <- assertSession pool (Sessions.changeVisibilityTimeout vtQuery)++    -- Verify message fields+    assertBool "messageId should be positive" (unMessageId (Pgmq.Types.messageId msg) > 0)+    assertEqual "messageId should match" msgId (Pgmq.Types.messageId msg)+    assertEqual "body should match" sentBody (unMessageBody (body msg))++    -- Visibility time should be ~60 seconds in the future+    let expectedMinVt = addUTCTime 55 beforeSetVt+    assertBool "visibilityTime should be in future" (visibilityTime msg >= expectedMinVt)++    cleanupQueue pool queueName++-- | Test batch set_vt decoder+testBatchSetVtDecoder :: Pool.Pool -> TestTree+testBatchSetVtDecoder p = testCase "batch set_vt returns correctly decoded messages" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send two messages+    let body1 = object ["batch_vt" .= (1 :: Int)]+    let body2 = object ["batch_vt" .= (2 :: Int)]+    msgId1 <- assertSession pool (Sessions.sendMessage (mkSendMessage queueName body1))+    msgId2 <- assertSession pool (Sessions.sendMessage (mkSendMessage queueName body2))++    -- Batch change visibility timeout+    let batchVtQuery =+          BatchVisibilityTimeoutQuery+            { queueName = queueName,+              messageIds = [msgId1, msgId2],+              visibilityTimeoutOffset = 45+            }+    messages <- assertSession pool (Sessions.batchChangeVisibilityTimeout batchVtQuery)++    assertEqual "Should return 2 messages" 2 (V.length messages)++    -- Verify each message is valid+    let msg1 = V.head messages+    let msg2 = messages V.! 1+    assertBool "first messageId should be positive" (unMessageId (Pgmq.Types.messageId msg1) > 0)+    assertBool "second messageId should be positive" (unMessageId (Pgmq.Types.messageId msg2) > 0)++    cleanupQueue pool queueName++-- | Test read_with_poll decoder+testReadWithPollDecoder :: Pool.Pool -> TestTree+testReadWithPollDecoder p = testCase "read_with_poll returns correctly decoded message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message first+    let sentBody = object ["function" .= ("read_with_poll" :: String)]+    _ <- assertSession pool (Sessions.sendMessage (mkSendMessage queueName sentBody))++    -- Read with poll (should return immediately since message exists)+    let pollQuery =+          ReadWithPollMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              maxPollSeconds = 1,+              pollIntervalMs = 100,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readWithPoll pollQuery)++    assertEqual "Should return 1 message" 1 (V.length messages)+    let msg = V.head messages+    assertValidMessage msg sentBody++    cleanupQueue pool queueName++-- Helper functions++mkSendMessage :: QueueName -> Value -> SendMessage+mkSendMessage qName bodyVal =+  SendMessage+    { queueName = qName,+      messageBody = MessageBody bodyVal,+      delay = Nothing+    }++-- | Assert that a message has valid fields+assertValidMessage :: Message -> Value -> IO ()+assertValidMessage msg expectedBody = do+  assertBool "messageId should be positive" (unMessageId (Pgmq.Types.messageId msg) > 0)+  assertBool "readCount should be >= 0" (readCount msg >= 0)+  assertEqual "body should match expected" expectedBody (unMessageBody (body msg))

@@ -0,0 +1,282 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Field-specific decoder validation tests+-- These tests verify each decoded field contains semantically correct values+-- This catches column swap bugs (e.g., reading read_ct into msg_id)+module DecoderValidationSpec (tests) where++import Data.Aeson (Value, object, (.=))+import Data.Time (addUTCTime, getCurrentTime)+import Data.Vector qualified as V+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Hasql.Session (Session)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( ReadMessage (..),+    SendMessage (..),+    SendMessageWithHeaders (..),+    VisibilityTimeoutQuery (..),+  )+import Pgmq.Types+  ( Message (..),+    MessageBody (..),+    MessageHeaders (..),+    QueueName,+    unMessageId,+  )+import Pgmq.Types qualified+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, testCase)+import TestUtils (assertSession, cleanupQueue)++-- | All decoder validation tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Decoder Field Validation"+    [ testMessageIdIsPositive p,+      testReadCountIncrements p,+      testEnqueuedAtBeforeRead p,+      testLastReadAtUpdates p,+      testVisibilityTimeAfterRead p,+      testBodyMatchesSent p,+      testHeadersMatchSent p,+      testNullHeadersWhenNotSent p+    ]++-- | Test that messageId is always positive+testMessageIdIsPositive :: Pool.Pool -> TestTree+testMessageIdIsPositive p = testCase "messageId is positive" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sendMsg = mkSendMessage queueName (object ["test" .= ("positive_id" :: String)])+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    -- Read message+    messages <- assertSession pool (mkReadMessage queueName)+    assertEqual "Should have 1 message" 1 (V.length messages)++    let msg = V.head messages+    assertBool+      "messageId should be positive (> 0)"+      (unMessageId (Pgmq.Types.messageId msg) > 0)+    cleanupQueue pool queueName++-- | Test that readCount increments on each read+testReadCountIncrements :: Pool.Pool -> TestTree+testReadCountIncrements p = testCase "readCount increments on each read" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sendMsg = mkSendMessage queueName (object ["test" .= ("read_count" :: String)])+    msgId <- assertSession pool (Sessions.sendMessage sendMsg)++    -- First read+    messages1 <- assertSession pool (mkReadMessage queueName)+    assertEqual "Should have 1 message" 1 (V.length messages1)+    let msg1 = V.head messages1+    assertEqual "readCount should be 1 after first read" 1 (readCount msg1)++    -- Make message visible again via set_vt+    let vtQuery =+          VisibilityTimeoutQuery+            { queueName = queueName,+              messageId = msgId,+              visibilityTimeoutOffset = 0 -- Make visible immediately+            }+    _ <- assertSession pool (Sessions.changeVisibilityTimeout vtQuery)++    -- Second read+    messages2 <- assertSession pool (mkReadMessage queueName)+    assertEqual "Should have 1 message" 1 (V.length messages2)+    let msg2 = V.head messages2+    assertEqual "readCount should be 2 after second read" 2 (readCount msg2)++    cleanupQueue pool queueName++-- | Test that enqueuedAt is before read time+testEnqueuedAtBeforeRead :: Pool.Pool -> TestTree+testEnqueuedAtBeforeRead p = testCase "enqueuedAt is before read time" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    beforeEnqueue <- getCurrentTime+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sendMsg = mkSendMessage queueName (object ["test" .= ("enqueued_at" :: String)])+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    afterEnqueue <- getCurrentTime++    -- Read message+    messages <- assertSession pool (mkReadMessage queueName)+    let msg = V.head messages++    -- enqueuedAt should be between beforeEnqueue and afterEnqueue+    assertBool+      "enqueuedAt should be after test start"+      (enqueuedAt msg >= beforeEnqueue)+    assertBool+      "enqueuedAt should be before read"+      (enqueuedAt msg <= afterEnqueue)++    cleanupQueue pool queueName++-- | Test that lastReadAt is set after reading+testLastReadAtUpdates :: Pool.Pool -> TestTree+testLastReadAtUpdates p = testCase "lastReadAt is set after reading" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sendMsg = mkSendMessage queueName (object ["test" .= ("last_read_at" :: String)])+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    beforeRead <- getCurrentTime++    -- First read+    messages1 <- assertSession pool (mkReadMessage queueName)+    let msg1 = V.head messages1++    afterRead <- getCurrentTime++    -- lastReadAt should be Just and between beforeRead and afterRead+    case lastReadAt msg1 of+      Nothing -> assertBool "lastReadAt should be set after first read" False+      Just lra -> do+        assertBool+          "lastReadAt should be >= beforeRead"+          (lra >= addUTCTime (-1) beforeRead) -- Allow 1 second tolerance+        assertBool+          "lastReadAt should be <= afterRead"+          (lra <= addUTCTime 1 afterRead) -- Allow 1 second tolerance+    cleanupQueue pool queueName++-- | Test that visibilityTime is in future after read with delay+testVisibilityTimeAfterRead :: Pool.Pool -> TestTree+testVisibilityTimeAfterRead p = testCase "visibilityTime is in future after read" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sendMsg = mkSendMessage queueName (object ["test" .= ("vt" :: String)])+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    beforeRead <- getCurrentTime++    -- Read message with 60 second visibility timeout+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 60, -- 60 second visibility timeout+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    let msg = V.head messages++    -- visibilityTime should be at least 55 seconds in the future+    let expectedMinVt = addUTCTime 55 beforeRead+    assertBool+      "visibilityTime should be at least 55 seconds in future"+      (visibilityTime msg >= expectedMinVt)++    cleanupQueue pool queueName++-- | Test that body exactly matches what was sent+testBodyMatchesSent :: Pool.Pool -> TestTree+testBodyMatchesSent p = testCase "body exactly matches sent message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message with specific body+    let sentBody =+          object+            [ "string_field" .= ("test value" :: String),+              "int_field" .= (42 :: Int),+              "bool_field" .= True,+              "nested" .= object ["inner" .= ("nested value" :: String)]+            ]+    let sendMsg = mkSendMessage queueName sentBody+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    -- Read message+    messages <- assertSession pool (mkReadMessage queueName)+    let msg = V.head messages+    let receivedBody = unMessageBody (Pgmq.Types.body msg)++    assertEqual "body should exactly match sent message" sentBody receivedBody++    cleanupQueue pool queueName++-- | Test that headers exactly match what was sent+testHeadersMatchSent :: Pool.Pool -> TestTree+testHeadersMatchSent p = testCase "headers exactly match sent message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message with headers+    let sentBody = object ["data" .= ("test" :: String)]+    let sentHeaders =+          object+            [ "trace_id" .= ("abc-123-xyz" :: String),+              "priority" .= (1 :: Int),+              "routing_key" .= ("queue.important" :: String)+            ]+    let sendMsg =+          SendMessageWithHeaders+            { queueName = queueName,+              messageBody = MessageBody sentBody,+              messageHeaders = MessageHeaders sentHeaders,+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.sendMessageWithHeaders sendMsg)++    -- Read message+    messages <- assertSession pool (mkReadMessage queueName)+    let msg = V.head messages++    assertEqual "headers should exactly match sent message" (Just sentHeaders) (Pgmq.Types.headers msg)++    cleanupQueue pool queueName++-- | Test that headers are null when not sent+testNullHeadersWhenNotSent :: Pool.Pool -> TestTree+testNullHeadersWhenNotSent p = testCase "headers are null when not sent" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)++    -- Send message without headers+    let sendMsg = mkSendMessage queueName (object ["test" .= ("no_headers" :: String)])+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    -- Read message+    messages <- assertSession pool (mkReadMessage queueName)+    let msg = V.head messages++    assertEqual "headers should be Nothing when not sent" Nothing (Pgmq.Types.headers msg)++    cleanupQueue pool queueName++-- Helper functions++mkSendMessage :: QueueName -> Value -> SendMessage+mkSendMessage qName bodyVal =+  SendMessage+    { queueName = qName,+      messageBody = MessageBody bodyVal,+      delay = Nothing+    }++mkReadMessage :: QueueName -> Session (V.Vector Message)+mkReadMessage qName =+  Sessions.readMessage+    ReadMessage+      { queueName = qName,+        delay = 30,+        batchSize = Just 1,+        conditional = Nothing+      }

@@ -0,0 +1,72 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Test database infrastructure using ephemeral-pg+module EphemeralDb+  ( -- * Database setup+    withPgmqDb,+    withPgmqPool,++    -- * Test fixtures+    TestFixture (..),+    withTestFixture,++    -- * Re-exports+    StartError,+  )+where++import Data.Text qualified as T+import Data.Word (Word32)+import EphemeralPg+  ( StartError,+    connectionSettings,+    withCached,+  )+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Pgmq.Migration qualified as Migration+import Pgmq.Types (QueueName, parseQueueName)+import System.Random (randomRIO)++-- | Run an action with a temporary PostgreSQL database that has pgmq schema installed+withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a)+withPgmqDb action = withCached $ \db -> do+  let connSettings = connectionSettings db+      poolConfig =+        PoolConfig.settings+          [ PoolConfig.size 3,+            PoolConfig.staticConnectionSettings connSettings+          ]+  pool <- Pool.acquire poolConfig+  -- Install pgmq schema+  installResult <- Pool.use pool Migration.migrate+  case installResult of+    Left poolErr -> error $ "Failed to install pgmq schema: " <> show poolErr+    Right (Left migrationErr) -> error $ "Migration failed: " <> show migrationErr+    Right (Right ()) -> action pool++-- | Run an action with a connection pool to a temporary PostgreSQL database+-- The database will have the pgmq schema installed+withPgmqPool :: (Pool.Pool -> IO a) -> IO (Either StartError a)+withPgmqPool = withPgmqDb++-- | Test fixture with isolated queue for a test+data TestFixture = TestFixture+  { pool :: !Pool.Pool,+    queueName :: !QueueName+  }++-- | Create an isolated test fixture with a random queue name+-- This allows tests to run in parallel without interfering with each other+withTestFixture :: Pool.Pool -> (TestFixture -> IO a) -> IO a+withTestFixture p action = do+  qName <- generateTestQueueName+  action TestFixture {pool = p, queueName = qName}++-- | Generate a random queue name for test isolation+generateTestQueueName :: IO QueueName+generateTestQueueName = do+  suffix <- randomRIO (10000 :: Word32, 99999)+  case parseQueueName ("test_queue_" <> T.pack (show suffix)) of+    Left err -> error $ "Failed to generate queue name: " <> show err+    Right name -> pure name

@@ -0,0 +1,111 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Hedgehog generators for pgmq types+module Generators+  ( genMessageBody,+    genMessageHeaders,+    genJsonValue,+    genJsonObject,+    genJsonScalar,+  )+where++import Data.Aeson (Value (..))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Scientific (fromFloatDigits)+import Data.Vector qualified as V+import Hedgehog (Gen)+import Hedgehog.Gen qualified as Gen+import Hedgehog.Range qualified as Range+import Pgmq.Types (MessageBody (..), MessageHeaders (..))++-- | Generate a random MessageBody containing a JSON object+genMessageBody :: Gen MessageBody+genMessageBody = MessageBody <$> genJsonObject++-- | Generate random MessageHeaders containing a JSON object+genMessageHeaders :: Gen MessageHeaders+genMessageHeaders = MessageHeaders <$> genJsonObject++-- | Generate a JSON object with random keys and values+genJsonObject :: Gen Value+genJsonObject = do+  numFields <- Gen.int (Range.linear 1 10)+  fields <- Gen.list (Range.singleton numFields) genField+  pure $ Object (KeyMap.fromList fields)+  where+    genField = do+      key <- genFieldName+      value <- genJsonValue+      pure (key, value)++    genFieldName = do+      len <- Gen.int (Range.linear 1 20)+      name <- Gen.text (Range.singleton len) Gen.alphaNum+      pure $ Key.fromText name++-- | Generate arbitrary JSON values (recursively)+genJsonValue :: Gen Value+genJsonValue =+  Gen.recursive+    Gen.choice+    -- Non-recursive cases+    [genJsonScalar]+    -- Recursive cases (less frequent due to shrinking)+    [ genJsonArray,+      genJsonObjectValue+    ]++-- | Generate scalar JSON values (non-recursive)+genJsonScalar :: Gen Value+genJsonScalar =+  Gen.choice+    [ genNull,+      genBool,+      genNumber,+      genString+    ]++genNull :: Gen Value+genNull = pure Null++genBool :: Gen Value+genBool = Bool <$> Gen.bool++genNumber :: Gen Value+genNumber =+  Gen.choice+    [ -- Integer-like numbers+      Number . fromIntegral <$> Gen.int (Range.linearFrom 0 (-1000000) 1000000),+      -- Floating point numbers+      Number . fromFloatDigits <$> Gen.double (Range.linearFracFrom 0 (-1000000) 1000000)+    ]++genString :: Gen Value+genString = do+  len <- Gen.int (Range.linear 0 100)+  txt <- Gen.text (Range.singleton len) Gen.unicode+  pure $ String txt++genJsonArray :: Gen Value+genJsonArray = do+  len <- Gen.int (Range.linear 0 5)+  elements <- Gen.list (Range.singleton len) genJsonScalar+  pure $ Array (V.fromList elements)++genJsonObjectValue :: Gen Value+genJsonObjectValue = do+  numFields <- Gen.int (Range.linear 1 5)+  fields <- Gen.list (Range.singleton numFields) genObjectField+  pure $ Object (KeyMap.fromList fields)+  where+    genObjectField = do+      key <- genKey+      value <- genJsonScalar+      pure (key, value)++    genKey = do+      len <- Gen.int (Range.linear 1 20)+      txt <- Gen.text (Range.singleton len) Gen.alphaNum+      pure $ Key.fromText txt

@@ -0,0 +1,33 @@+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import AdvancedOpsSpec qualified+import AllFunctionsDecoderSpec qualified+import DecoderValidationSpec qualified+import EphemeralDb (withPgmqPool)+import MessageSpec qualified+import QueueSpec qualified+import RoundTripSpec qualified+import SchemaSpec qualified+import Test.Tasty (defaultMain, testGroup)++main :: IO ()+main = do+  -- Run tests with a shared temporary database+  result <- withPgmqPool $ \pool -> do+    let tree =+          testGroup+            "pgmq-hasql"+            [ QueueSpec.tests pool,+              MessageSpec.tests pool,+              AdvancedOpsSpec.tests pool,+              SchemaSpec.tests pool,+              RoundTripSpec.tests pool,+              DecoderValidationSpec.tests pool,+              AllFunctionsDecoderSpec.tests pool+            ]+    defaultMain tree+  case result of+    Left err -> error $ "Failed to start temp database: " <> show err+    Right () -> pure ()

@@ -0,0 +1,300 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests for message operations+module MessageSpec (tests) where++import Data.Aeson (object, (.=))+import Data.Vector qualified as V+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( BatchSendMessage (..),+    BatchSendMessageWithHeaders (..),+    MessageQuery (..),+    ReadMessage (..),+    SendMessage (..),+    SendMessageWithHeaders (..),+    VisibilityTimeoutQuery (..),+  )+import Pgmq.Types (MessageBody (..), MessageHeaders (..), MessageId (..))+import Pgmq.Types qualified as PgmqTypes+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, testCase)+import TestUtils (assertSession, cleanupQueue)++-- | All message operation tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Message Operations"+    [ testSendMessageNullDelay p,+      testSendMessageZeroDelay p,+      testBatchSendMessageNullDelay p,+      testReadMessage p,+      testDeleteMessage p,+      testArchiveMessage p,+      testChangeVisibilityTimeout p,+      testSendMessageWithHeaders p,+      testBatchSendMessageWithHeaders p+    ]++-- | Test sendMessage with delay=Nothing (validates COALESCE fix)+-- This is a critical test - it validates that our fix for NULL delay+-- correctly uses the pgmq function overload resolution+testSendMessageNullDelay :: Pool.Pool -> TestTree+testSendMessageNullDelay p = testCase "sendMessage with delay=Nothing works" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let msg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["test" .= ("hello" :: String)]),+              delay = Nothing+            }+    msgId <- assertSession pool (Sessions.sendMessage msg)+    assertBool "Message ID should be positive" (unMessageId msgId > 0)+    -- Verify message can be read immediately (no delay)+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should read 1 message" 1 (V.length messages)+    cleanupQueue pool queueName++-- | Test sendMessage with delay=Just 0+testSendMessageZeroDelay :: Pool.Pool -> TestTree+testSendMessageZeroDelay p = testCase "sendMessage with delay=Just 0 works" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let msg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["test" .= ("hello" :: String)]),+              delay = Just 0+            }+    msgId <- assertSession pool (Sessions.sendMessage msg)+    assertBool "Message ID should be positive" (unMessageId msgId > 0)+    -- Verify message can be read immediately+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should read 1 message" 1 (V.length messages)+    cleanupQueue pool queueName++-- | Test batchSendMessage with delay=Nothing (validates COALESCE fix)+testBatchSendMessageNullDelay :: Pool.Pool -> TestTree+testBatchSendMessageNullDelay p = testCase "batchSendMessage with delay=Nothing works" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let msgs =+          BatchSendMessage+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["batch" .= (1 :: Int)]),+                  MessageBody (object ["batch" .= (2 :: Int)]),+                  MessageBody (object ["batch" .= (3 :: Int)])+                ],+              delay = Nothing+            }+    msgIds <- assertSession pool (Sessions.batchSendMessage msgs)+    assertEqual "Should return 3 message IDs" 3 (length msgIds)+    -- Verify messages can be read+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 10,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should read 3 messages" 3 (V.length messages)+    cleanupQueue pool queueName++-- | Test readMessage+testReadMessage :: Pool.Pool -> TestTree+testReadMessage p = testCase "readMessage retrieves and hides message" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send a message+    let msg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["key" .= ("value" :: String)]),+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.sendMessage msg)+    -- Read the message+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30, -- 30 second visibility timeout+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should read 1 message" 1 (V.length messages)+    -- Message should not be readable again (still hidden)+    messages2 <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should not read hidden message" 0 (V.length messages2)+    cleanupQueue pool queueName++-- | Test deleteMessage+testDeleteMessage :: Pool.Pool -> TestTree+testDeleteMessage p = testCase "deleteMessage removes message from queue" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send and read a message+    let sendMsg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["delete" .= ("me" :: String)]),+              delay = Nothing+            }+    msgId <- assertSession pool (Sessions.sendMessage sendMsg)+    -- Delete the message+    let deleteQuery =+          MessageQuery+            { queueName = queueName,+              messageId = msgId+            }+    deleted <- assertSession pool (Sessions.deleteMessage deleteQuery)+    assertBool "Delete should succeed" deleted+    -- Verify message can't be read anymore+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Queue should be empty after delete" 0 (V.length messages)+    cleanupQueue pool queueName++-- | Test archiveMessage+testArchiveMessage :: Pool.Pool -> TestTree+testArchiveMessage p = testCase "archiveMessage moves message to archive" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send a message+    let sendMsg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["archive" .= ("me" :: String)]),+              delay = Nothing+            }+    msgId <- assertSession pool (Sessions.sendMessage sendMsg)+    -- Archive the message+    let archiveQuery =+          MessageQuery+            { queueName = queueName,+              messageId = msgId+            }+    archived <- assertSession pool (Sessions.archiveMessage archiveQuery)+    assertBool "Archive should succeed" archived+    -- Verify message can't be read from queue anymore+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Queue should be empty after archive" 0 (V.length messages)+    cleanupQueue pool queueName++-- | Test changeVisibilityTimeout+testChangeVisibilityTimeout :: Pool.Pool -> TestTree+testChangeVisibilityTimeout p = testCase "changeVisibilityTimeout extends/resets VT" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    -- Send a message+    let sendMsg =+          SendMessage+            { queueName = queueName,+              messageBody = MessageBody (object ["vt" .= ("test" :: String)]),+              delay = Nothing+            }+    msgId <- assertSession pool (Sessions.sendMessage sendMsg)+    -- Read the message to set visibility timeout+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    _ <- assertSession pool (Sessions.readMessage readQuery)+    -- Change visibility timeout+    let vtQuery =+          VisibilityTimeoutQuery+            { queueName = queueName,+              messageId = msgId,+              visibilityTimeoutOffset = 60+            }+    msg <- assertSession pool (Sessions.changeVisibilityTimeout vtQuery)+    assertEqual "Should return the message" msgId (PgmqTypes.messageId msg)+    cleanupQueue pool queueName++-- | Test sendMessageWithHeaders+testSendMessageWithHeaders :: Pool.Pool -> TestTree+testSendMessageWithHeaders p = testCase "sendMessageWithHeaders includes headers" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let msg =+          SendMessageWithHeaders+            { queueName = queueName,+              messageBody = MessageBody (object ["data" .= ("test" :: String)]),+              messageHeaders = MessageHeaders (object ["trace_id" .= ("abc123" :: String)]),+              delay = Nothing+            }+    msgId <- assertSession pool (Sessions.sendMessageWithHeaders msg)+    assertBool "Message ID should be positive" (unMessageId msgId > 0)+    -- Read and verify headers+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    assertEqual "Should read 1 message" 1 (V.length messages)+    let readMsg = V.head messages+    case PgmqTypes.headers readMsg of+      Just _ -> assertBool "Headers should contain trace_id" True+      Nothing -> assertBool "Headers should be present" False+    cleanupQueue pool queueName++-- | Test batchSendMessageWithHeaders+testBatchSendMessageWithHeaders :: Pool.Pool -> TestTree+testBatchSendMessageWithHeaders p = testCase "batchSendMessageWithHeaders sends batch with headers" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    assertSession pool (Sessions.createQueue queueName)+    let msgs =+          BatchSendMessageWithHeaders+            { queueName = queueName,+              messageBodies =+                [ MessageBody (object ["batch" .= (1 :: Int)]),+                  MessageBody (object ["batch" .= (2 :: Int)])+                ],+              messageHeaders =+                [ MessageHeaders (object ["idx" .= (1 :: Int)]),+                  MessageHeaders (object ["idx" .= (2 :: Int)])+                ],+              delay = Nothing+            }+    msgIds <- assertSession pool (Sessions.batchSendMessageWithHeaders msgs)+    assertEqual "Should return 2 message IDs" 2 (length msgIds)+    cleanupQueue pool queueName

@@ -0,0 +1,96 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests for queue metrics operations+module MetricsSpec (tests) where++import Data.Aeson (object, (.=))+import Data.Text (Text)+import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types (QueueMetrics (..))+import Pgmq.Hasql.Statements.Types qualified as Types+import Pgmq.Types (MessageBody (..), QueueName, parseQueueName, queueNameToText)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, testCase, (@?=))+import TestUtils (assertRight, assertSession, cleanupQueue)++-- | All metrics tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Metrics"+    [ testQueueMetrics p,+      testQueueMetricsEmpty p,+      testAllQueueMetrics p+    ]++-- | Test queueMetrics returns correct metrics+testQueueMetrics :: Pool.Pool -> TestTree+testQueueMetrics p = testCase "queueMetrics returns queue statistics" $ do+  withTestFixture p $ \TestFixture {pool, queueName = qName} -> do+    assertSession pool (Sessions.createQueue qName)+    -- Send some messages+    let msg = makeSendMessage qName+    _ <- assertSession pool (Sessions.sendMessage msg)+    _ <- assertSession pool (Sessions.sendMessage msg)+    _ <- assertSession pool (Sessions.sendMessage msg)+    -- Get metrics+    metrics <- assertSession pool (Sessions.queueMetrics qName)+    -- Verify metrics+    queueLength metrics @?= 3+    totalMessages metrics @?= 3+    queueVisibleLength metrics @?= 3+    assertEqual "Queue name should match" (queueNameToText qName) (metricsQueueName metrics)+    cleanupQueue pool qName++-- | Test queueMetrics for empty queue+testQueueMetricsEmpty :: Pool.Pool -> TestTree+testQueueMetricsEmpty p = testCase "queueMetrics works for empty queue" $ do+  withTestFixture p $ \TestFixture {pool, queueName = qName} -> do+    assertSession pool (Sessions.createQueue qName)+    -- Get metrics for empty queue+    metrics <- assertSession pool (Sessions.queueMetrics qName)+    queueLength metrics @?= 0+    totalMessages metrics @?= 0+    queueVisibleLength metrics @?= 0+    cleanupQueue pool qName++-- | Test allQueueMetrics returns all queues+testAllQueueMetrics :: Pool.Pool -> TestTree+testAllQueueMetrics p = testCase "allQueueMetrics returns metrics for all queues" $ do+  queueName1 <- assertRight $ parseQueueName "test_metrics_q1"+  queueName2 <- assertRight $ parseQueueName "test_metrics_q2"+  -- Create two queues+  assertSession p (Sessions.createQueue queueName1)+  assertSession p (Sessions.createQueue queueName2)+  -- Send a message to the first queue+  let msg = makeSendMessage queueName1+  _ <- assertSession p (Sessions.sendMessage msg)+  -- Get all metrics+  allMetrics <- assertSession p Sessions.allQueueMetrics+  -- Verify both queues are in the results+  let queueNames = map metricsQueueName allMetrics+  assertBool "Queue 1 should be in metrics" (queueNameToText queueName1 `elem` queueNames)+  assertBool "Queue 2 should be in metrics" (queueNameToText queueName2 `elem` queueNames)+  -- Verify queue 1 has the correct message count+  let q1Metrics = filter (\m -> metricsQueueName m == queueNameToText queueName1) allMetrics+  case q1Metrics of+    [m] -> queueLength m @?= 1+    _ -> assertBool "Queue 1 should have metrics" False+  -- Cleanup+  cleanupQueue p queueName1+  cleanupQueue p queueName2++-- | Helper to extract queue name from metrics (avoids ambiguity)+metricsQueueName :: QueueMetrics -> Text+metricsQueueName QueueMetrics {queueName = qn} = qn++-- | Helper to create SendMessage (avoids field ambiguity)+makeSendMessage :: QueueName -> Types.SendMessage+makeSendMessage qName =+  Types.SendMessage+    { Types.queueName = qName,+      Types.messageBody = MessageBody (object ["test" .= (1 :: Int)]),+      Types.delay = Nothing+    }

@@ -0,0 +1,88 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Tests for queue management operations+module QueueSpec (tests) where++import EphemeralDb (TestFixture (..), withTestFixture)+import Hasql.Pool qualified as Pool+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Types (Queue (..), parseQueueName)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, testCase, (@?=))+import TestUtils+  ( assertRight,+    assertSession,+    cleanupQueue,+  )++-- | All queue management tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Queue Management"+    [ testCreateQueue p,+      testDropQueue p,+      testDropNonExistentQueue p,+      testListQueues p,+      testCreateUnloggedQueue p+      -- Note: testCreatePartitionedQueue is skipped because it requires pg_partman extension+    ]++testCreateQueue :: Pool.Pool -> TestTree+testCreateQueue p = testCase "createQueue creates a new queue" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    -- Create the queue+    assertSession pool (Sessions.createQueue queueName)+    -- Verify it exists by listing queues+    queues <- assertSession pool Sessions.listQueues+    let queueNames = map (\q -> name q) queues+    assertBool "Queue should be in list" (queueName `elem` queueNames)+    -- Cleanup+    cleanupQueue pool queueName++testDropQueue :: Pool.Pool -> TestTree+testDropQueue p = testCase "dropQueue removes an existing queue" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    -- Create then drop the queue+    assertSession pool (Sessions.createQueue queueName)+    dropped <- assertSession pool (Sessions.dropQueue queueName)+    dropped @?= True+    -- Verify it's gone+    queues <- assertSession pool Sessions.listQueues+    let queueNames = map (\q -> name q) queues+    assertBool "Queue should not be in list" (queueName `notElem` queueNames)++testDropNonExistentQueue :: Pool.Pool -> TestTree+testDropNonExistentQueue p = testCase "dropQueue returns False for non-existent queue" $ do+  withTestFixture p $ \TestFixture {pool, queueName} -> do+    -- Try to drop a queue that doesn't exist+    dropped <- assertSession pool (Sessions.dropQueue queueName)+    dropped @?= False++testListQueues :: Pool.Pool -> TestTree+testListQueues p = testCase "listQueues returns all created queues" $ do+  queueName1 <- assertRight $ parseQueueName "test_list_q1"+  queueName2 <- assertRight $ parseQueueName "test_list_q2"+  -- Create two queues+  assertSession p (Sessions.createQueue queueName1)+  assertSession p (Sessions.createQueue queueName2)+  -- List and verify both exist+  queues <- assertSession p Sessions.listQueues+  let queueNames = map (\q -> name q) queues+  assertBool "Queue 1 should be in list" (queueName1 `elem` queueNames)+  assertBool "Queue 2 should be in list" (queueName2 `elem` queueNames)+  -- Cleanup+  cleanupQueue p queueName1+  cleanupQueue p queueName2++testCreateUnloggedQueue :: Pool.Pool -> TestTree+testCreateUnloggedQueue p = testCase "createUnloggedQueue creates an unlogged queue" $ do+  qName <- assertRight $ parseQueueName "test_unlogged_q"+  assertSession p (Sessions.createUnloggedQueue qName)+  -- Verify it exists+  queues <- assertSession p Sessions.listQueues+  let matchingQueues = filter (\q -> name q == qName) queues+  -- Just verify the queue was created (isUnlogged status depends on pgmq schema version)+  assertBool "Queue should exist" (not (null matchingQueues))+  -- Cleanup+  cleanupQueue p qName

@@ -0,0 +1,144 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Property-based round-trip tests+-- Verifies that data survives encode -> database -> decode cycles+module RoundTripSpec (tests) where++import Data.Vector qualified as V+import EphemeralDb (TestFixture (..), withTestFixture)+import Generators (genMessageBody, genMessageHeaders)+import Hasql.Pool qualified as Pool+import Hedgehog (annotateShow, forAll, (===))+import Hedgehog qualified as H+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types+  ( ReadMessage (..),+    SendMessage (..),+    SendMessageWithHeaders (..),+  )+import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), unMessageId)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.Hedgehog (testProperty)+import TestUtils (assertSession, cleanupQueue)++-- | All round-trip property tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Round-Trip Properties"+    [ testProperty "message body survives round-trip" (propMessageBodyRoundTrip p),+      testProperty "message with headers survives round-trip" (propMessageWithHeadersRoundTrip p),+      testProperty "timestamp ordering is preserved" (propTimestampOrdering p)+    ]++-- | Property: message body survives send/read round-trip+propMessageBodyRoundTrip :: Pool.Pool -> H.Property+propMessageBodyRoundTrip p = H.property $ do+  body <- forAll genMessageBody+  result <- H.evalIO $ withTestFixture p $ \TestFixture {pool, queueName} -> do+    -- Create queue+    assertSession pool (Sessions.createQueue queueName)++    -- Send message with random body+    let sendMsg =+          SendMessage+            { queueName = queueName,+              messageBody = body,+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    -- Read message back+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    cleanupQueue pool queueName+    pure messages++  -- Verify body is preserved+  annotateShow result+  H.assert (V.length result == 1)+  let msg = V.head result+  unMessageBody (body :: MessageBody) === unMessageBody (Pgmq.Types.body msg)++-- | Property: message with headers survives round-trip+propMessageWithHeadersRoundTrip :: Pool.Pool -> H.Property+propMessageWithHeadersRoundTrip p = H.property $ do+  body <- forAll genMessageBody+  hdrs <- forAll genMessageHeaders+  result <- H.evalIO $ withTestFixture p $ \TestFixture {pool, queueName} -> do+    -- Create queue+    assertSession pool (Sessions.createQueue queueName)++    -- Send message with headers+    let sendMsg =+          SendMessageWithHeaders+            { queueName = queueName,+              messageBody = body,+              messageHeaders = hdrs,+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.sendMessageWithHeaders sendMsg)++    -- Read message back+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30,+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    cleanupQueue pool queueName+    pure messages++  -- Verify body and headers are preserved+  annotateShow result+  H.assert (V.length result == 1)+  let msg = V.head result+  unMessageBody (body :: MessageBody) === unMessageBody (Pgmq.Types.body msg)+  Just (unMessageHeaders hdrs) === Pgmq.Types.headers msg++-- | Property: timestamps maintain ordering (enqueued_at <= vt when read)+propTimestampOrdering :: Pool.Pool -> H.Property+propTimestampOrdering p = H.property $ do+  body <- forAll genMessageBody+  result <- H.evalIO $ withTestFixture p $ \TestFixture {pool, queueName} -> do+    -- Create queue+    assertSession pool (Sessions.createQueue queueName)++    -- Send message+    let sendMsg =+          SendMessage+            { queueName = queueName,+              messageBody = body,+              delay = Nothing+            }+    _ <- assertSession pool (Sessions.sendMessage sendMsg)++    -- Read message with visibility timeout+    let readQuery =+          ReadMessage+            { queueName = queueName,+              delay = 30, -- 30 second visibility timeout+              batchSize = Just 1,+              conditional = Nothing+            }+    messages <- assertSession pool (Sessions.readMessage readQuery)+    cleanupQueue pool queueName+    pure messages++  -- Verify timestamp ordering+  H.assert (V.length result == 1)+  let msg = V.head result+  -- enqueued_at should be before visibility_time (vt is set to future after read)+  H.assert (enqueuedAt msg < visibilityTime msg)+  -- Message ID should be positive+  H.assert (unMessageId (messageId msg) > 0)+  -- Read count should be 1 after first read+  readCount msg === 1

@@ -0,0 +1,91 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}++-- | Schema introspection tests+-- Verifies that pgmq.message_record type matches decoder expectations+module SchemaSpec (tests) where++import Data.Text (Text)+import Data.Vector qualified as V+import Hasql.Decoders qualified as D+import Hasql.Pool qualified as Pool+import Hasql.Session (Session, statement)+import Hasql.Statement (Statement, preparable)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertEqual, testCase)+import TestUtils (assertSession)++-- | All schema introspection tests+tests :: Pool.Pool -> TestTree+tests p =+  testGroup+    "Schema Introspection"+    [ testMessageRecordColumnOrder p,+      testMessageRecordColumnTypes p+    ]++-- | Expected column order for pgmq.message_record+-- This must match the order in Pgmq.Hasql.Decoders.messageDecoder+expectedMessageRecordColumns :: [(Text, Text)]+expectedMessageRecordColumns =+  [ ("msg_id", "int8"),+    ("read_ct", "int4"),+    ("enqueued_at", "timestamptz"),+    ("last_read_at", "timestamptz"),+    ("vt", "timestamptz"),+    ("message", "jsonb"),+    ("headers", "jsonb")+  ]++-- | Query to get column order of pgmq.message_record type+-- Returns columns in their attribute order+messageRecordColumnsQuery :: Statement () (V.Vector (Text, Text))+messageRecordColumnsQuery =+  preparable+    sql+    mempty+    decoder+  where+    sql =+      "SELECT a.attname::text, t.typname::text \+      \FROM pg_type typ \+      \JOIN pg_namespace ns ON typ.typnamespace = ns.oid \+      \JOIN pg_attribute a ON a.attrelid = typ.typrelid \+      \JOIN pg_type t ON a.atttypid = t.oid \+      \WHERE ns.nspname = 'pgmq' \+      \  AND typ.typname = 'message_record' \+      \  AND a.attnum > 0 \+      \ORDER BY a.attnum"++    decoder :: D.Result (V.Vector (Text, Text))+    decoder = D.rowVector $ (,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text)++-- | Query columns of pgmq.message_record+getMessageRecordColumns :: Session (V.Vector (Text, Text))+getMessageRecordColumns = statement () messageRecordColumnsQuery++-- | Test that pgmq.message_record columns match expected order+-- This catches schema drift where columns are reordered or renamed+testMessageRecordColumnOrder :: Pool.Pool -> TestTree+testMessageRecordColumnOrder p = testCase "message_record column order matches decoder" $ do+  columns <- assertSession p getMessageRecordColumns+  let actualColumns = V.toList columns+      expectedColNames = map fst expectedMessageRecordColumns+      actualColNames = map fst actualColumns++  assertEqual+    "Column order must match decoder expectations (msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers)"+    expectedColNames+    actualColNames++-- | Test that pgmq.message_record column types match expected types+-- This catches type changes that would cause decoding failures+testMessageRecordColumnTypes :: Pool.Pool -> TestTree+testMessageRecordColumnTypes p = testCase "message_record column types match decoder" $ do+  columns <- assertSession p getMessageRecordColumns+  let actualColumns = V.toList columns++  assertEqual+    "Column types must match decoder expectations"+    expectedMessageRecordColumns+    actualColumns

@@ -0,0 +1,59 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Common test utilities for pgmq-hasql tests+module TestUtils+  ( -- * Session helpers+    runSession,+    assertSession,+    assertSessionFails,++    -- * Queue helpers+    cleanupQueue,++    -- * Assertion helpers+    assertRight,+    assertJust,+  )+where++import Hasql.Pool qualified as Pool+import Hasql.Session (Session)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Types (QueueName)+import Test.Tasty.HUnit (assertFailure)++-- | Run a session against a pool+runSession :: Pool.Pool -> Session a -> IO (Either Pool.UsageError a)+runSession = Pool.use++-- | Run a session and fail the test if it errors+assertSession :: Pool.Pool -> Session a -> IO a+assertSession p session = do+  result <- runSession p session+  case result of+    Left err -> assertFailure $ "Session failed: " <> show err+    Right a -> pure a++-- | Assert that a session fails+assertSessionFails :: Pool.Pool -> Session a -> IO ()+assertSessionFails p session = do+  result <- runSession p session+  case result of+    Left _ -> pure ()+    Right _ -> assertFailure "Expected session to fail but it succeeded"++-- | Clean up a queue by dropping it (ignores errors)+cleanupQueue :: Pool.Pool -> QueueName -> IO ()+cleanupQueue p qName = do+  _ <- runSession p (Sessions.dropQueue qName)+  pure ()++-- | Assert that an Either is Right and return the value+assertRight :: (Show e) => Either e a -> IO a+assertRight (Left err) = assertFailure $ "Expected Right but got Left: " <> show err+assertRight (Right a) = pure a++-- | Assert that a Maybe is Just and return the value+assertJust :: Maybe a -> IO a+assertJust Nothing = assertFailure "Expected Just but got Nothing"+assertJust (Just a) = pure a