diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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 diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,20 @@ +Copyright (c) 2025 Nadeem Bitar + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/pgmq-hasql.cabal b/pgmq-hasql.cabal new file mode 100644 --- /dev/null +++ b/pgmq-hasql.cabal @@ -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 diff --git a/src/Pgmq.hs b/src/Pgmq.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq.hs @@ -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, + ) diff --git a/src/Pgmq/Hasql/Decoders.hs b/src/Pgmq/Hasql/Decoders.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Decoders.hs @@ -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+) diff --git a/src/Pgmq/Hasql/Encoders.hs b/src/Pgmq/Hasql/Encoders.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Encoders.hs @@ -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)) diff --git a/src/Pgmq/Hasql/Prelude.hs b/src/Pgmq/Hasql/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Prelude.hs @@ -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) diff --git a/src/Pgmq/Hasql/Quasi.hs b/src/Pgmq/Hasql/Quasi.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Quasi.hs @@ -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 diff --git a/src/Pgmq/Hasql/Sessions.hs b/src/Pgmq/Hasql/Sessions.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Sessions.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements.hs b/src/Pgmq/Hasql/Statements.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Statements.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements/Message.hs b/src/Pgmq/Hasql/Statements/Message.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Statements/Message.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements/QueueManagement.hs b/src/Pgmq/Hasql/Statements/QueueManagement.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Statements/QueueManagement.hs @@ -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_ +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()" diff --git a/src/Pgmq/Hasql/Statements/QueueObservability.hs b/src/Pgmq/Hasql/Statements/QueueObservability.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Statements/QueueObservability.hs @@ -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 diff --git a/src/Pgmq/Hasql/Statements/Types.hs b/src/Pgmq/Hasql/Statements/Types.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Hasql/Statements/Types.hs @@ -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) diff --git a/test/AdvancedOpsSpec.hs b/test/AdvancedOpsSpec.hs new file mode 100644 --- /dev/null +++ b/test/AdvancedOpsSpec.hs @@ -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 diff --git a/test/AllFunctionsDecoderSpec.hs b/test/AllFunctionsDecoderSpec.hs new file mode 100644 --- /dev/null +++ b/test/AllFunctionsDecoderSpec.hs @@ -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)) diff --git a/test/DecoderValidationSpec.hs b/test/DecoderValidationSpec.hs new file mode 100644 --- /dev/null +++ b/test/DecoderValidationSpec.hs @@ -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 + } diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs new file mode 100644 --- /dev/null +++ b/test/EphemeralDb.hs @@ -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 diff --git a/test/Generators.hs b/test/Generators.hs new file mode 100644 --- /dev/null +++ b/test/Generators.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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 () diff --git a/test/MessageSpec.hs b/test/MessageSpec.hs new file mode 100644 --- /dev/null +++ b/test/MessageSpec.hs @@ -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 diff --git a/test/MetricsSpec.hs b/test/MetricsSpec.hs new file mode 100644 --- /dev/null +++ b/test/MetricsSpec.hs @@ -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 + } diff --git a/test/QueueSpec.hs b/test/QueueSpec.hs new file mode 100644 --- /dev/null +++ b/test/QueueSpec.hs @@ -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 diff --git a/test/RoundTripSpec.hs b/test/RoundTripSpec.hs new file mode 100644 --- /dev/null +++ b/test/RoundTripSpec.hs @@ -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 diff --git a/test/SchemaSpec.hs b/test/SchemaSpec.hs new file mode 100644 --- /dev/null +++ b/test/SchemaSpec.hs @@ -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 diff --git a/test/TestUtils.hs b/test/TestUtils.hs new file mode 100644 --- /dev/null +++ b/test/TestUtils.hs @@ -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