pgmq-config (empty) → 0.1.3.0

raw patch · 9 files changed

+851/−0 lines, 9 filesdep +basedep +containersdep +effectful-core

Dependencies added: base, containers, effectful-core, ephemeral-pg, generic-lens, hasql, hasql-pool, lens, pgmq-config, pgmq-core, pgmq-effectful, pgmq-hasql, pgmq-migration, random, tasty, tasty-hunit, text

Files

• CHANGELOG.md +10/−0
• LICENSE +20/−0
• pgmq-config.cabal +104/−0
• src/Pgmq/Config.hs +188/−0
• src/Pgmq/Config/Effectful.hs +159/−0
• src/Pgmq/Config/Types.hs +109/−0
• test/ConfigSpec.hs +204/−0
• test/EphemeralDb.hs +37/−0
• test/Main.hs +20/−0

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

@@ -0,0 +1,20 @@+Copyright (c) 2025 Nadeem Bitar++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

@@ -0,0 +1,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

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

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

@@ -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])

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

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

@@ -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 ()