[PGMQ] implement advisory lock when initializing pgmq

Otherwise, concurrent calls to 'IF NOT EXISTS' when creating extension
could result in unique constraint errors.

src/Database/PGMQ/Simple.hs

@@ -52,7 +52,17 @@ import Safe (headMay)
 {-| Initialize PGMQ given a PostgreSQL connection. Mainly concerned
     with creating the 'pgmq' extension. -}
 initialize :: PSQL.Connection -> IO ()
-initialize conn = void $ PSQL.execute_ conn [sql| CREATE EXTENSION IF NOT EXISTS pgmq |]
+initialize conn =
+  -- OK so this is a bit tricky because of the usage of IF NOT EXISTS:
+  -- https://stackoverflow.com/questions/29900845/create-schema-if-not-exists-raises-duplicate-key-error
+  -- PostgreSQL will complain badly if 'initialize' is called
+  -- from multiple threads at once.
+  -- Hence, we use 'pg_advisory_xact_lock' to lock ourselves
+  -- out of this situation.
+  PSQL.withTransaction conn $ do
+    let magicLockId = 1122334455 :: Int
+    _ <- PSQL.query conn [sql| SELECT pg_advisory_xact_lock(?) |] (PSQL.Only magicLockId) :: IO [PSQL.Only ()]
+    void $ PSQL.execute_ conn [sql| CREATE EXTENSION IF NOT EXISTS pgmq |]
 
 
 {-| Archives message in given queue for given id

tests/Test/Integration/Broker.hs

@@ -40,8 +40,7 @@ instance FromJSON Message where
 
 pgmqBrokerInitParams :: IO (BT.BrokerInitParams PGMQBroker Message)
 pgmqBrokerInitParams = do
-  connectInfo <- getEnvConnectInfo
-  pure $ PGMQBrokerInitParams connectInfo
+  PGMQBrokerInitParams <$> getEnvConnectInfo
 
 
 withBroker :: (BT.HasBroker b Message)

tests/Test/Integration/Worker.hs

@@ -71,6 +71,7 @@ data SimpleException = SimpleException String
 instance Exception SimpleException
 
 
+-- | Perform Action for this worker
 pa :: (HasWorkerBroker b Message) => State b a -> BT.BrokerMessage b (Job Message) -> IO ()
 pa _state bm = do
   let job' = BT.toA $ BT.getMessage bm
@@ -366,5 +367,4 @@ millisecond = 1000
 
 pgmqWorkerBrokerInitParams :: IO (BT.BrokerInitParams PGMQBroker (Job Message))
 pgmqWorkerBrokerInitParams = do
-  connectInfo <- getEnvConnectInfo
-  pure $ PGMQBrokerInitParams connectInfo
+  PGMQBrokerInitParams <$> getEnvConnectInfo