diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,10 @@ +# Revision history for pgmq-config + +## 0.1.3.0 -- 2026-03-12 + +* Initial release +* Declarative queue configuration DSL (QueueConfig, QueueType, etc.) +* Smart constructors: standardQueue, unloggedQueue, partitionedQueue +* Modifiers: withNotifyInsert, withFifoIndex, withTopicBinding +* Reconciliation: ensureQueues, ensureQueuesWithPool, ensureQueuesReport +* Optional effectful integration (behind flag) 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-config.cabal b/pgmq-config.cabal new file mode 100644 --- /dev/null +++ b/pgmq-config.cabal @@ -0,0 +1,104 @@ +cabal-version: 3.4 +name: pgmq-config +version: 0.1.3.0 +synopsis: + Declarative queue configuration for PGMQ (PostgreSQL Message Queue) + +description: + A declarative DSL for configuring pgmq queues. Define your queue + topology as Haskell values and call a single function at startup + to ensure all queues exist with the desired settings. Supports + standard, unlogged, and partitioned queues, notification setup, + FIFO indexes, and topic bindings. All operations are idempotent. + +homepage: https://github.com/shinzui/pgmq-hs +license: MIT +license-file: LICENSE +author: Nadeem Bitar +maintainer: Nadeem Bitar +category: Database +build-type: Simple +extra-doc-files: CHANGELOG.md + +flag effectful + description: Enable effectful integration + default: True + manual: False + +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.Config + Pgmq.Config.Types + + default-extensions: + DeriveGeneric + DuplicateRecordFields + GeneralisedNewtypeDeriving + ImportQualifiedPost + NamedFieldPuns + OverloadedLabels + OverloadedStrings + + build-depends: + , base >=4.18 && <5 + , containers >=0.6 && <0.8 + , generic-lens ^>=2.2 || ^>=2.3 + , hasql ^>=1.10 + , hasql-pool ^>=1.4 + , lens ^>=5.3 + , pgmq-core >=0.1.3 && <0.2 + , pgmq-hasql >=0.1.3 && <0.2 + , text ^>=2.1 + + if flag(effectful) + exposed-modules: Pgmq.Config.Effectful + build-depends: + , effectful-core ^>=2.5 || ^>=2.6 + , pgmq-effectful >=0.1.3 && <0.2 + + hs-source-dirs: src + default-language: GHC2024 + +test-suite pgmq-config-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: + ConfigSpec + EphemeralDb + + default-extensions: + DeriveGeneric + DuplicateRecordFields + GeneralisedNewtypeDeriving + ImportQualifiedPost + NamedFieldPuns + OverloadedLabels + OverloadedStrings + + build-depends: + , base >=4.18 && <5 + , ephemeral-pg >=0.2.1 + , generic-lens ^>=2.2 || ^>=2.3 + , hasql + , hasql-pool ^>=1.4 + , lens ^>=5.3 + , pgmq-config + , pgmq-core + , pgmq-hasql + , pgmq-migration + , random ^>=1.2 + , tasty ^>=1.5 + , tasty-hunit ^>=0.10 + , text diff --git a/src/Pgmq/Config.hs b/src/Pgmq/Config.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Config.hs @@ -0,0 +1,188 @@ +module Pgmq.Config + ( -- * Queue Configuration Types + QueueConfig (..), + QueueType (..), + PartitionConfig (..), + NotifyConfig (..), + + -- * Smart Constructors + standardQueue, + unloggedQueue, + partitionedQueue, + + -- * Modifiers + withNotifyInsert, + withFifoIndex, + withTopicBinding, + + -- * Reconciliation + ensureQueues, + ensureQueuesWithPool, + + -- * Reconciliation with Report + ReconcileAction (..), + ensureQueuesReport, + ) +where + +import Control.Lens ((^.)) +import Data.Foldable (for_) +import Data.Generics.Labels () +import Data.Set qualified as Set +import Data.Text qualified as T +import Hasql.Pool qualified as Pool +import Hasql.Session (Session) +import Pgmq.Config.Types +import Pgmq.Hasql.Sessions qualified as Sessions +import Pgmq.Hasql.Statements.Types qualified as StmtTypes +import Pgmq.Types + ( QueueName, + TopicPattern, + queueNameToText, + topicPatternToText, + ) + +-- | Ensure all declared queues exist with the desired settings. +-- This is idempotent — safe to call on every application startup. +-- Operations are additive only: queues not in the config are left untouched. +ensureQueues :: [QueueConfig] -> Session () +ensureQueues configs = + for_ configs applyQueueConfig + +-- | Convenience wrapper that runs 'ensureQueues' against a connection pool. +ensureQueuesWithPool :: Pool.Pool -> [QueueConfig] -> IO (Either Pool.UsageError ()) +ensureQueuesWithPool pool configs = + Pool.use pool (ensureQueues configs) + +-- | Like 'ensureQueues', but returns a report of actions taken. +-- Queries existing state first and skips operations that are already satisfied. +ensureQueuesReport :: [QueueConfig] -> Session [ReconcileAction] +ensureQueuesReport configs = do + existingQueues <- Sessions.listQueues + existingBindings <- Sessions.listTopicBindings + existingThrottles <- Sessions.listNotifyInsertThrottles + + let existingQueueNames = Set.fromList (map (\q -> q ^. #name) existingQueues) + existingBindingSet = + Set.fromList + [ (b ^. #bindingQueueName, topicPatternToText (b ^. #bindingPattern)) + | b <- existingBindings + ] + existingNotifySet = Set.fromList (map (\t -> t ^. #throttleQueueName) existingThrottles) + + concat <$> traverse (reconcileQueue existingQueueNames existingBindingSet existingNotifySet) configs + +-- | Apply a single queue config without checking existing state. +applyQueueConfig :: QueueConfig -> Session () +applyQueueConfig cfg = do + let qn = cfg ^. #queueName + -- Create the queue + case cfg ^. #queueType of + StandardQueue -> + Sessions.createQueue qn + UnloggedQueue -> + Sessions.createUnloggedQueue qn + PartitionedQueue pc -> + Sessions.createPartitionedQueue + StmtTypes.CreatePartitionedQueue + { queueName = qn, + partitionInterval = pc ^. #partitionInterval, + retentionInterval = pc ^. #retentionInterval + } + + -- Enable notifications if configured + for_ (cfg ^. #notifyInsert) $ \nc -> + Sessions.enableNotifyInsert + StmtTypes.EnableNotifyInsert + { queueName = qn, + throttleIntervalMs = nc ^. #throttleMs + } + + -- Create FIFO index if requested + if cfg ^. #fifoIndex + then Sessions.createFifoIndex qn + else pure () + + -- Bind topic patterns + for_ (cfg ^. #topicBindings) $ \pat -> + Sessions.bindTopic + StmtTypes.BindTopic + { topicPattern = pat, + queueName = qn + } + +-- | Reconcile a single queue config against existing state, returning actions taken. +reconcileQueue :: + Set.Set QueueName -> + Set.Set (T.Text, T.Text) -> + Set.Set T.Text -> + QueueConfig -> + Session [ReconcileAction] +reconcileQueue existingQueues existingBindings existingNotify cfg = do + let qn = cfg ^. #queueName + qnText = queueNameToText qn + + -- Queue creation + queueAction <- + if Set.member qn existingQueues + then pure [SkippedQueue qn] + else do + case cfg ^. #queueType of + StandardQueue -> + Sessions.createQueue qn + UnloggedQueue -> + Sessions.createUnloggedQueue qn + PartitionedQueue pc -> + Sessions.createPartitionedQueue + StmtTypes.CreatePartitionedQueue + { queueName = qn, + partitionInterval = pc ^. #partitionInterval, + retentionInterval = pc ^. #retentionInterval + } + pure [CreatedQueue qn (cfg ^. #queueType)] + + -- Notification + notifyAction <- case cfg ^. #notifyInsert of + Nothing -> pure [] + Just nc -> + if Set.member qnText existingNotify + then pure [SkippedNotify qn] + else do + Sessions.enableNotifyInsert + StmtTypes.EnableNotifyInsert + { queueName = qn, + throttleIntervalMs = nc ^. #throttleMs + } + pure [EnabledNotify qn (nc ^. #throttleMs)] + + -- FIFO index — no way to query if index exists, so always apply (idempotent) + fifoAction <- + if cfg ^. #fifoIndex + then do + Sessions.createFifoIndex qn + pure [CreatedFifoIndex qn] + else pure [] + + -- Topic bindings + bindingActions <- concat <$> traverse (reconcileBinding qn qnText existingBindings) (cfg ^. #topicBindings) + + pure (queueAction ++ notifyAction ++ fifoAction ++ bindingActions) + +-- | Reconcile a single topic binding. +reconcileBinding :: + QueueName -> + T.Text -> + Set.Set (T.Text, T.Text) -> + TopicPattern -> + Session [ReconcileAction] +reconcileBinding qn qnText existingBindings pat = + let patText = topicPatternToText pat + in if Set.member (qnText, patText) existingBindings + then pure [SkippedTopicBinding qn pat] + else do + Sessions.bindTopic + StmtTypes.BindTopic + { topicPattern = pat, + queueName = qn + } + pure [BoundTopic qn pat] diff --git a/src/Pgmq/Config/Effectful.hs b/src/Pgmq/Config/Effectful.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Config/Effectful.hs @@ -0,0 +1,159 @@ +module Pgmq.Config.Effectful + ( -- * Reconciliation + ensureQueuesEff, + + -- * Reconciliation with Report + ensureQueuesReportEff, + ) +where + +import Control.Lens ((^.)) +import Data.Generics.Labels () +import Data.Set qualified as Set +import Data.Text qualified as T +import Effectful (Eff, (:>)) +import Pgmq.Config.Types +import Pgmq.Effectful.Effect qualified as Eff +import Pgmq.Hasql.Statements.Types qualified as StmtTypes +import Pgmq.Types + ( QueueName, + TopicPattern, + queueNameToText, + topicPatternToText, + ) + +-- | Ensure all declared queues exist using the Pgmq effect. +ensureQueuesEff :: (Eff.Pgmq :> es) => [QueueConfig] -> Eff es () +ensureQueuesEff configs = + mapM_ applyQueueConfigEff configs + +-- | Like 'ensureQueuesEff', but returns a report of actions taken. +ensureQueuesReportEff :: (Eff.Pgmq :> es) => [QueueConfig] -> Eff es [ReconcileAction] +ensureQueuesReportEff configs = do + existingQueues <- Eff.listQueues + existingBindings <- Eff.listTopicBindings + existingThrottles <- Eff.listNotifyInsertThrottles + + let existingQueueNames = Set.fromList (map (\q -> q ^. #name) existingQueues) + existingBindingSet = + Set.fromList + [ (b ^. #bindingQueueName, topicPatternToText (b ^. #bindingPattern)) + | b <- existingBindings + ] + existingNotifySet = Set.fromList (map (\t -> t ^. #throttleQueueName) existingThrottles) + + concat <$> traverse (reconcileQueueEff existingQueueNames existingBindingSet existingNotifySet) configs + +-- | Apply a single queue config without checking existing state. +applyQueueConfigEff :: (Eff.Pgmq :> es) => QueueConfig -> Eff es () +applyQueueConfigEff cfg = do + let qn = cfg ^. #queueName + case cfg ^. #queueType of + StandardQueue -> + Eff.createQueue qn + UnloggedQueue -> + Eff.createUnloggedQueue qn + PartitionedQueue pc -> + Eff.createPartitionedQueue + StmtTypes.CreatePartitionedQueue + { queueName = qn, + partitionInterval = pc ^. #partitionInterval, + retentionInterval = pc ^. #retentionInterval + } + + case cfg ^. #notifyInsert of + Nothing -> pure () + Just nc -> + Eff.enableNotifyInsert + StmtTypes.EnableNotifyInsert + { queueName = qn, + throttleIntervalMs = nc ^. #throttleMs + } + + if cfg ^. #fifoIndex + then Eff.createFifoIndex qn + else pure () + + mapM_ + ( \pat -> + Eff.bindTopic + StmtTypes.BindTopic + { topicPattern = pat, + queueName = qn + } + ) + (cfg ^. #topicBindings) + +-- | Reconcile a single queue config against existing state. +reconcileQueueEff :: + (Eff.Pgmq :> es) => + Set.Set QueueName -> + Set.Set (T.Text, T.Text) -> + Set.Set T.Text -> + QueueConfig -> + Eff es [ReconcileAction] +reconcileQueueEff existingQueues existingBindings existingNotify cfg = do + let qn = cfg ^. #queueName + qnText = queueNameToText qn + + queueAction <- + if Set.member qn existingQueues + then pure [SkippedQueue qn] + else do + case cfg ^. #queueType of + StandardQueue -> + Eff.createQueue qn + UnloggedQueue -> + Eff.createUnloggedQueue qn + PartitionedQueue pc -> + Eff.createPartitionedQueue + StmtTypes.CreatePartitionedQueue + { queueName = qn, + partitionInterval = pc ^. #partitionInterval, + retentionInterval = pc ^. #retentionInterval + } + pure [CreatedQueue qn (cfg ^. #queueType)] + + notifyAction <- case cfg ^. #notifyInsert of + Nothing -> pure [] + Just nc -> + if Set.member qnText existingNotify + then pure [SkippedNotify qn] + else do + Eff.enableNotifyInsert + StmtTypes.EnableNotifyInsert + { queueName = qn, + throttleIntervalMs = nc ^. #throttleMs + } + pure [EnabledNotify qn (nc ^. #throttleMs)] + + fifoAction <- + if cfg ^. #fifoIndex + then do + Eff.createFifoIndex qn + pure [CreatedFifoIndex qn] + else pure [] + + bindingActions <- concat <$> traverse (reconcileBindingEff qn qnText existingBindings) (cfg ^. #topicBindings) + + pure (queueAction ++ notifyAction ++ fifoAction ++ bindingActions) + +-- | Reconcile a single topic binding. +reconcileBindingEff :: + (Eff.Pgmq :> es) => + QueueName -> + T.Text -> + Set.Set (T.Text, T.Text) -> + TopicPattern -> + Eff es [ReconcileAction] +reconcileBindingEff qn qnText existingBindings pat = + let patText = topicPatternToText pat + in if Set.member (qnText, patText) existingBindings + then pure [SkippedTopicBinding qn pat] + else do + Eff.bindTopic + StmtTypes.BindTopic + { topicPattern = pat, + queueName = qn + } + pure [BoundTopic qn pat] diff --git a/src/Pgmq/Config/Types.hs b/src/Pgmq/Config/Types.hs new file mode 100644 --- /dev/null +++ b/src/Pgmq/Config/Types.hs @@ -0,0 +1,109 @@ +module Pgmq.Config.Types + ( -- * Queue Configuration + QueueConfig (..), + QueueType (..), + PartitionConfig (..), + NotifyConfig (..), + + -- * Smart Constructors + standardQueue, + unloggedQueue, + partitionedQueue, + + -- * Modifiers + withNotifyInsert, + withFifoIndex, + withTopicBinding, + + -- * Reconciliation Report + ReconcileAction (..), + ) +where + +import Control.Lens ((%~), (&)) +import Data.Generics.Labels () +import Data.Int (Int32) +import Data.Text (Text) +import GHC.Generics (Generic) +import Pgmq.Types (QueueName, TopicPattern) + +-- | Describes the desired state of a single pgmq queue. +data QueueConfig = QueueConfig + { queueName :: !QueueName, + queueType :: !QueueType, + notifyInsert :: !(Maybe NotifyConfig), + fifoIndex :: !Bool, + topicBindings :: ![TopicPattern] + } + deriving stock (Generic, Show) + +-- | The type of queue to create. +data QueueType + = -- | A standard queue with write-ahead logging. + StandardQueue + | -- | An unlogged queue — faster writes, but data is lost on crash. + UnloggedQueue + | -- | A partitioned queue for high-throughput scenarios. + PartitionedQueue !PartitionConfig + deriving stock (Show) + +-- | Configuration for a partitioned queue. +data PartitionConfig = PartitionConfig + { partitionInterval :: !Text, + retentionInterval :: !Text + } + deriving stock (Generic, Show) + +-- | Configuration for insert notifications (LISTEN/NOTIFY). +data NotifyConfig = NotifyConfig + { -- | Minimum milliseconds between notifications. Nothing uses pgmq default (250ms). + throttleMs :: !(Maybe Int32) + } + deriving stock (Generic, Show) + +-- | An action taken (or skipped) during queue reconciliation. +data ReconcileAction + = CreatedQueue !QueueName !QueueType + | EnabledNotify !QueueName !(Maybe Int32) + | CreatedFifoIndex !QueueName + | BoundTopic !QueueName !TopicPattern + | SkippedQueue !QueueName + | SkippedNotify !QueueName + | SkippedFifoIndex !QueueName + | SkippedTopicBinding !QueueName !TopicPattern + deriving stock (Show) + +-- | Create a standard queue configuration with no extras. +standardQueue :: QueueName -> QueueConfig +standardQueue qn = + QueueConfig + { queueName = qn, + queueType = StandardQueue, + notifyInsert = Nothing, + fifoIndex = False, + topicBindings = [] + } + +-- | Create an unlogged queue configuration (faster, no WAL, lost on crash). +unloggedQueue :: QueueName -> QueueConfig +unloggedQueue qn = + (standardQueue qn) {queueType = UnloggedQueue} + +-- | Create a partitioned queue configuration. +partitionedQueue :: QueueName -> PartitionConfig -> QueueConfig +partitionedQueue qn pc = + (standardQueue qn) {queueType = PartitionedQueue pc} + +-- | Enable LISTEN/NOTIFY on message insert. +withNotifyInsert :: Maybe Int32 -> QueueConfig -> QueueConfig +withNotifyInsert ms cfg = + cfg {notifyInsert = Just NotifyConfig {throttleMs = ms}} + +-- | Add a FIFO index for strict message ordering. +withFifoIndex :: QueueConfig -> QueueConfig +withFifoIndex cfg = cfg {fifoIndex = True} + +-- | Bind a topic pattern for AMQP-style routing. +withTopicBinding :: TopicPattern -> QueueConfig -> QueueConfig +withTopicBinding pat cfg = + cfg & #topicBindings %~ (++ [pat]) diff --git a/test/ConfigSpec.hs b/test/ConfigSpec.hs new file mode 100644 --- /dev/null +++ b/test/ConfigSpec.hs @@ -0,0 +1,204 @@ +{-# LANGUAGE OverloadedStrings #-} + +module ConfigSpec + ( tests, + ) +where + +import Control.Lens ((^.)) +import Data.Generics.Labels () +import Data.Text qualified as T +import Data.Word (Word32) +import Hasql.Pool qualified as Pool +import Hasql.Session qualified +import Pgmq.Config +import Pgmq.Hasql.Sessions qualified as Sessions +import Pgmq.Types (QueueName, parseQueueName, parseTopicPattern, queueNameToText, topicPatternToText) +import System.Random (randomRIO) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=)) + +tests :: Pool.Pool -> TestTree +tests pool = + testGroup + "Pgmq.Config" + [ testEnsureQueuesCreatesStandard pool, + testEnsureQueuesCreatesUnlogged pool, + testEnsureQueuesIdempotent pool, + testEnsureQueuesIncremental pool, + testEnsureQueuesWithNotify pool, + testEnsureQueuesWithFifo pool, + testEnsureQueuesWithTopicBinding pool + ] + +-- | Helper to run a session and fail on error +runSession :: Pool.Pool -> Hasql.Session.Session a -> IO a +runSession p session = do + result <- Pool.use p session + case result of + Left err -> assertFailure $ "Session failed: " <> show err + Right a -> pure a + +-- | Generate a random queue name for test isolation +genQueueName :: IO QueueName +genQueueName = do + suffix <- randomRIO (10000 :: Word32, 99999) + case parseQueueName ("cfg_test_" <> T.pack (show suffix)) of + Left err -> error $ "Failed to generate queue name: " <> show err + Right qn -> pure qn + +-- | Clean up a queue (ignore errors) +cleanupQueue :: Pool.Pool -> QueueName -> IO () +cleanupQueue p qn = do + _ <- Pool.use p (Sessions.dropQueue qn) + pure () + +testEnsureQueuesCreatesStandard :: Pool.Pool -> TestTree +testEnsureQueuesCreatesStandard pool = testCase "creates a standard queue" $ do + qn <- genQueueName + let configs = [standardQueue qn] + actions <- runSession pool (ensureQueuesReport configs) + assertBool "should have CreatedQueue action" $ + any isCreatedQueue actions + -- Verify queue exists + queues <- runSession pool Sessions.listQueues + assertBool "queue should be in list" $ + any (\q -> (q ^. #name) == qn) queues + cleanupQueue pool qn + +testEnsureQueuesCreatesUnlogged :: Pool.Pool -> TestTree +testEnsureQueuesCreatesUnlogged pool = testCase "creates an unlogged queue" $ do + qn <- genQueueName + let configs = [unloggedQueue qn] + actions <- runSession pool (ensureQueuesReport configs) + assertBool "should have CreatedQueue action" $ + any isCreatedQueue actions + -- Verify it's unlogged + queues <- runSession pool Sessions.listQueues + let mq = filter (\q -> (q ^. #name) == qn) queues + case mq of + [q] -> (q ^. #isUnlogged) @?= True + _ -> assertFailure "queue not found" + cleanupQueue pool qn + +testEnsureQueuesIdempotent :: Pool.Pool -> TestTree +testEnsureQueuesIdempotent pool = testCase "is idempotent (second run skips)" $ do + qn <- genQueueName + let configs = [standardQueue qn] + -- First run: creates + actions1 <- runSession pool (ensureQueuesReport configs) + assertBool "first run should create" $ + any isCreatedQueue actions1 + -- Second run: skips + actions2 <- runSession pool (ensureQueuesReport configs) + assertBool "second run should skip" $ + all isSkipped actions2 + cleanupQueue pool qn + +testEnsureQueuesIncremental :: Pool.Pool -> TestTree +testEnsureQueuesIncremental pool = testCase "creates only new queues incrementally" $ do + qn1 <- genQueueName + qn2 <- genQueueName + -- First run: create one queue + _ <- runSession pool (ensureQueuesReport [standardQueue qn1]) + -- Second run: add a second queue + actions <- runSession pool (ensureQueuesReport [standardQueue qn1, standardQueue qn2]) + -- qn1 should be skipped, qn2 should be created + let qn1Actions = filter (actionForQueue qn1) actions + qn2Actions = filter (actionForQueue qn2) actions + assertBool "existing queue should be skipped" $ + all isSkipped qn1Actions + assertBool "new queue should be created" $ + any isCreatedQueue qn2Actions + cleanupQueue pool qn1 + cleanupQueue pool qn2 + +testEnsureQueuesWithNotify :: Pool.Pool -> TestTree +testEnsureQueuesWithNotify pool = testCase "enables notify insert" $ do + qn <- genQueueName + let configs = [withNotifyInsert (Just 500) (standardQueue qn)] + actions <- runSession pool (ensureQueuesReport configs) + assertBool "should have EnabledNotify action" $ + any isEnabledNotify actions + -- Second run should skip notify + actions2 <- runSession pool (ensureQueuesReport configs) + assertBool "second run should skip notify" $ + any isSkippedNotify actions2 + cleanupQueue pool qn + +testEnsureQueuesWithFifo :: Pool.Pool -> TestTree +testEnsureQueuesWithFifo pool = testCase "creates FIFO index" $ do + qn <- genQueueName + let configs = [withFifoIndex (standardQueue qn)] + actions <- runSession pool (ensureQueuesReport configs) + assertBool "should have CreatedFifoIndex action" $ + any isFifoIndex actions + cleanupQueue pool qn + +testEnsureQueuesWithTopicBinding :: Pool.Pool -> TestTree +testEnsureQueuesWithTopicBinding pool = testCase "binds topic pattern" $ do + qn <- genQueueName + pat <- case parseTopicPattern "orders.*" of + Left err -> assertFailure ("Failed to parse topic pattern: " <> show err) >> error "unreachable" + Right p -> pure p + let configs = [withTopicBinding pat (standardQueue qn)] + actions <- runSession pool (ensureQueuesReport configs) + assertBool "should have BoundTopic action" $ + any isBoundTopic actions + -- Verify binding exists + bindings <- runSession pool Sessions.listTopicBindings + assertBool "binding should exist" $ + any + ( \b -> + (b ^. #bindingQueueName) == queueNameToText qn + && topicPatternToText (b ^. #bindingPattern) == "orders.*" + ) + bindings + -- Second run should skip + actions2 <- runSession pool (ensureQueuesReport configs) + assertBool "second run should skip topic binding" $ + any isSkippedTopicBinding actions2 + cleanupQueue pool qn + +-- Helpers for checking action types + +isCreatedQueue :: ReconcileAction -> Bool +isCreatedQueue (CreatedQueue _ _) = True +isCreatedQueue _ = False + +isSkipped :: ReconcileAction -> Bool +isSkipped (SkippedQueue _) = True +isSkipped (SkippedNotify _) = True +isSkipped (SkippedFifoIndex _) = True +isSkipped (SkippedTopicBinding _ _) = True +isSkipped _ = False + +isSkippedTopicBinding :: ReconcileAction -> Bool +isSkippedTopicBinding (SkippedTopicBinding _ _) = True +isSkippedTopicBinding _ = False + +isEnabledNotify :: ReconcileAction -> Bool +isEnabledNotify (EnabledNotify _ _) = True +isEnabledNotify _ = False + +isSkippedNotify :: ReconcileAction -> Bool +isSkippedNotify (SkippedNotify _) = True +isSkippedNotify _ = False + +isFifoIndex :: ReconcileAction -> Bool +isFifoIndex (CreatedFifoIndex _) = True +isFifoIndex _ = False + +isBoundTopic :: ReconcileAction -> Bool +isBoundTopic (BoundTopic _ _) = True +isBoundTopic _ = False + +actionForQueue :: QueueName -> ReconcileAction -> Bool +actionForQueue qn (CreatedQueue q _) = q == qn +actionForQueue qn (SkippedQueue q) = q == qn +actionForQueue qn (EnabledNotify q _) = q == qn +actionForQueue qn (SkippedNotify q) = q == qn +actionForQueue qn (CreatedFifoIndex q) = q == qn +actionForQueue qn (SkippedFifoIndex q) = q == qn +actionForQueue qn (BoundTopic q _) = q == qn +actionForQueue qn (SkippedTopicBinding q _) = q == qn diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs new file mode 100644 --- /dev/null +++ b/test/EphemeralDb.hs @@ -0,0 +1,37 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | Test database infrastructure using ephemeral-pg +module EphemeralDb + ( -- * Database setup + withPgmqDb, + + -- * Re-exports + StartError, + ) +where + +import EphemeralPg + ( StartError, + connectionSettings, + withCached, + ) +import Hasql.Pool qualified as Pool +import Hasql.Pool.Config qualified as PoolConfig +import Pgmq.Migration qualified as Migration + +-- | 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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,20 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Main (main) where + +import ConfigSpec qualified +import EphemeralDb (withPgmqDb) +import Test.Tasty (defaultMain, testGroup) + +main :: IO () +main = do + result <- withPgmqDb $ \pool -> do + let tree = + testGroup + "pgmq-config" + [ ConfigSpec.tests pool + ] + defaultMain tree + case result of + Left err -> error $ "Failed to start temp database: " <> show err + Right () -> pure ()