[refactor] split into smaller components

haskell-bee, haskell-bee-tests, haskell-bee-pgmq, haskell-bee-redis,
haskell-bee-stm

TODO.md

+# TODO
+
+- [ ] implement XMPP broker
+- [ ] implement Amazon SQS broker
+- [ ] demonstrate/test fan out (i.e. a task creating other tasks)
+- [ ] implement tests for delayed message
+- [ ] split pgmq/redis/stm implementations into separate sub-packages

cabal.project

@@ -3,12 +3,15 @@ index-state: 2023-12-10T10:34:46Z
 with-compiler: ghc-9.4.8
 
 packages:
-    ./
+    haskell-bee/
+    haskell-bee-tests/
+    haskell-bee-pgmq/
+    haskell-bee-redis/
+    haskell-bee-stm/
+
+tests: true
 
 source-repository-package
     type: git
     location: https://gitlab.iscpif.fr/gargantext/haskell-pgmq
     tag: 1dd92f0aa8e9f8096064e5656c336e562680f4e3
-    
-tests: true
-

demo/CHANGELOG.md

+# Revision history for haskell-bee-demo
+
+## 0.1.0.0 -- YYYY-mm-dd
+
+* First version. Released on an unsuspecting world.

demo/haskell-bee-demo.cabal

@@ -32,7 +32,7 @@ version:            0.1.0.0
 license:            AGPL-3.0-or-later
 
 -- The file containing the license text.
-license-file:       LICENSE
+license-file:       ../LICENSE
 
 -- The package author(s).
 author:             Przemysław Kaminski

haskell-bee.cabal → haskell-bee-pgmq/haskell-bee-pgmq.cabal

@@ -11,7 +11,7 @@ cabal-version:      3.4
 --   http://haskell.org/cabal/users-guide/
 --
 -- The name of the package.
-name:               haskell-bee
+name:               haskell-bee-pgmq
 
 -- The package version.
 -- See the Haskell package versioning policy (PVP) for standards
@@ -35,7 +35,7 @@ homepage:           https://gitlab.iscpif.fr/gargantext/haskell-bee
 license:            AGPL-3.0-or-later
 
 -- The file containing the license text.
-license-file:       LICENSE
+license-file:       ../LICENSE
 
 -- The package author(s).
 author:             Gargantext
@@ -63,13 +63,7 @@ library
     import:           warnings
 
     -- Modules exported by the library.
-    exposed-modules:  Async.Worker
-                    , Async.Worker.Broker
-                    , Async.Worker.Broker.PGMQ
-                    , Async.Worker.Broker.Redis
-                    , Async.Worker.Broker.STM
-                    , Async.Worker.Broker.Types
-                    , Async.Worker.Types
+    exposed-modules:  Async.Worker.Broker.PGMQ
 
     -- Modules included in this library but not exported.
     -- other-modules:
@@ -84,8 +78,6 @@ library
                     , containers >= 0.6.7 && < 0.8
                     , deepseq >= 1.0.0.0 && < 1.7
                     , haskell-pgmq >= 0.1.0.0 && < 0.2
-                    , hedis >= 0.15.2 && < 0.16
-                    , mtl >= 2.2 && < 2.4
                     , postgresql-libpq >= 0.10 && < 0.11
                     , postgresql-simple >= 0.6 && < 0.8
                     , safe >= 0.3 && < 0.4
@@ -97,6 +89,9 @@ library
                     , units >= 2.4 && < 2.5
                     , unix-time >= 0.4.11 && < 0.5
 
+
+                    , haskell-bee
+
     -- Directories containing source files.
     hs-source-dirs:   src
 
@@ -124,6 +119,7 @@ executable simple-worker
                     , text >= 1.2 && < 2.2
                     
                     , haskell-bee
+                    , haskell-bee-pgmq
 
     -- Directories containing source files.
     hs-source-dirs:   bin/simple-worker
@@ -142,52 +138,17 @@ executable simple-worker
 
     ghc-options: -threaded
 
-test-suite test-unit
-    -- Import common warning flags.
-    import:           warnings
-    
-    type:             exitcode-stdio-1.0
-    
-    build-depends:    base ^>=4.17.2.0
-                    , aeson >= 2.1 && < 2.3
-                    , tasty >= 1.5 && < 1.6
-                    , tasty-hunit >= 0.10 && < 0.11
-                    , tasty-quickcheck >= 0.10 && < 0.12
-                    , unix-time >= 0.4.11 && < 0.5
-                    
-                    , haskell-bee
-
-    -- Directories containing source files.
-    hs-source-dirs:   tests
-    main-is: unit-tests.hs
-
-    -- Base language which the package is written in.
-    default-language: Haskell2010
-
-    default-extensions:
-      DuplicateRecordFields
-      GeneralizedNewtypeDeriving
-      ImportQualifiedPost
-      NamedFieldPuns
-      OverloadedStrings
-      RecordWildCards
-
-    ghc-options: -threaded
-
-test-suite test-integration
+test-suite pgmq-test-integration
     -- Import common warning flags.
     import:           warnings
     
     type:             exitcode-stdio-1.0
 
-    other-modules:  Test.Integration.Broker
-                  , Test.Integration.Utils
-                  , Test.Integration.Worker
+    other-modules:  Test.Integration.PGMQ
         
     build-depends:    base ^>=4.17.2.0
                     , aeson >= 2.1 && < 2.3
                     , containers >= 0.6 && < 0.8
-                    , hedis >= 0.15.2 && < 0.16
                     , hspec >= 2.11 && < 3
                     , postgresql-simple >= 0.6 && < 0.8
                     , random-strings == 0.1.1.0
@@ -197,6 +158,8 @@ test-suite test-integration
                     , text >= 1.2 && < 2.2
                     
                     , haskell-bee
+                    , haskell-bee-tests
+                    , haskell-bee-pgmq
 
     -- Directories containing source files.
     hs-source-dirs:   tests

haskell-bee-pgmq/tests/Test/Integration/PGMQ.hs

+module Test.Integration.PGMQ
+ ( pgmqBrokerInitParams
+ , pgmqWorkerBrokerInitParams )
+where
+
+import Async.Worker.Broker.PGMQ qualified as PGMQ
+import Async.Worker.Broker.Types qualified as BT
+import Async.Worker.Types (Job(..))
+import Data.Maybe (fromMaybe)
+import Database.PostgreSQL.Simple qualified as PSQL
+import System.Environment (lookupEnv)
+import Test.Integration.Broker qualified as TIB
+import Test.Integration.Worker qualified as TIW
+
+-- | Visibility timeout is a very important parameter for PGMQ. It is
+-- mainly used when reading a job: it specifies for how many seconds
+-- this job should be invisible for other workers. We need more tests
+-- and setting this correctly, preferably in accordance with
+-- 'Job.timeout'. Issue is that at the broker level we don't know
+-- anything about 'Job'...
+--
+-- The lower the value, the more probable that some other worker will
+-- pick up the same job at about the same time (before broker marks it
+-- as invisible).
+defaultPGMQVt :: Int
+defaultPGMQVt = 1
+
+-- | PSQL connect info that is fetched from env
+getPSQLEnvConnectInfo :: IO PSQL.ConnectInfo
+getPSQLEnvConnectInfo = do
+  pgUser <- lookupEnv "POSTGRES_USER"
+  pgDb <- lookupEnv "POSTGRES_DB"
+  pgPass <- lookupEnv "POSTGRES_PASSWORD"
+  pgHost <- lookupEnv "POSTGRES_HOST"
+  -- https://hackage.haskell.org/package/postgresql-simple-0.7.0.0/docs/Database-PostgreSQL-Simple.html#t:ConnectInfo
+  pure $ PSQL.defaultConnectInfo { PSQL.connectUser = fromMaybe "postgres" pgUser
+                                 , PSQL.connectDatabase = fromMaybe "postgres" pgDb
+                                 , PSQL.connectHost = fromMaybe "localhost" pgHost
+                                 , PSQL.connectPassword = fromMaybe "postgres" pgPass }
+
+
+pgmqBrokerInitParams :: IO (BT.BrokerInitParams PGMQ.PGMQBroker TIB.Message)
+pgmqBrokerInitParams = do
+  conn <- getPSQLEnvConnectInfo
+  return $ PGMQ.PGMQBrokerInitParams conn defaultPGMQVt
+
+
+
+
+pgmqWorkerBrokerInitParams :: IO (BT.BrokerInitParams PGMQ.PGMQBroker (Job TIW.Message))
+pgmqWorkerBrokerInitParams = do
+  conn <- getPSQLEnvConnectInfo
+  return $ PGMQ.PGMQBrokerInitParams conn defaultPGMQVt

haskell-bee-pgmq/tests/integration-tests.hs

+module Main where
+
+import Test.Integration.Broker (brokerTests)
+import Test.Integration.PGMQ (pgmqBrokerInitParams, pgmqWorkerBrokerInitParams)
+import Test.Integration.Worker (workerTests, multiWorkerTests)
+import Test.Tasty
+import Test.Tasty.Hspec
+
+
+
+main :: IO ()
+main = do
+  pgmqBInitParams <- pgmqBrokerInitParams
+  pgmqBrokerSpec <- testSpec "brokerTests" (brokerTests pgmqBInitParams)
+  pgmqWBInitParams <- pgmqWorkerBrokerInitParams
+  pgmqWorkerSpec <- testSpec "workerTests" (workerTests pgmqWBInitParams)
+  pgmqMultiWorkerSpec <- testSpec "multiWorkerTests" (multiWorkerTests pgmqWBInitParams 5)
+
+  defaultMain $ testGroup "PGMQ integration tests"
+    [
+      pgmqBrokerSpec
+    , pgmqWorkerSpec
+    , pgmqMultiWorkerSpec
+   ]
+

haskell-bee-redis/haskell-bee-redis.cabal

+cabal-version:      3.4
+-- The cabal-version field refers to the version of the .cabal specification,
+-- and can be different from the cabal-install (the tool) version and the
+-- Cabal (the library) version you are using. As such, the Cabal (the library)
+-- version used must be equal or greater than the version stated in this field.
+-- Starting from the specification version 2.2, the cabal-version field must be
+-- the first thing in the cabal file.
+
+-- Initial package description 'haskell-pgmq' generated by
+-- 'cabal init'. For further documentation, see:
+--   http://haskell.org/cabal/users-guide/
+--
+-- The name of the package.
+name:               haskell-bee-redis
+
+-- The package version.
+-- See the Haskell package versioning policy (PVP) for standards
+-- guiding when and how versions should be incremented.
+-- https://pvp.haskell.org
+-- PVP summary:     +-+------- breaking API changes
+--                  | | +----- non-breaking API additions
+--                  | | | +--- code changes with no API change
+version:            0.1.0.0
+
+-- A short (one-line) description of the package.
+-- synopsis:
+
+-- A longer description of the package.
+-- description:
+
+-- URL for the project homepage or repository.
+homepage:           https://gitlab.iscpif.fr/gargantext/haskell-bee
+
+-- The license under which the package is released.
+license:            AGPL-3.0-or-later
+
+-- The file containing the license text.
+license-file:       ../LICENSE
+
+-- The package author(s).
+author:             Gargantext
+
+-- An email address to which users can send suggestions, bug reports, and patches.
+maintainer:         gargantext@iscpif.fr
+
+-- A copyright notice.
+-- copyright:
+category:           Concurrency
+build-type:         Simple
+
+-- Extra doc files to be distributed with the package, such as a CHANGELOG or a README.
+extra-doc-files:    CHANGELOG.md
+                  , README.md
+
+-- Extra source files to be distributed with the package, such as examples, or a tutorial module.
+-- extra-source-files:
+
+common warnings
+    ghc-options: -Wall
+
+library
+    -- Import common warning flags.
+    import:           warnings
+
+    -- Modules exported by the library.
+    exposed-modules:  Async.Worker.Broker.Redis
+
+    -- Modules included in this library but not exported.
+    -- other-modules:
+
+    -- LANGUAGE extensions used by modules in this package.
+    -- other-extensions:
+
+    -- Other library packages from which modules are imported.
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , bytestring >= 0.11 && < 0.13
+                    , containers >= 0.6.7 && < 0.8
+                    , deepseq >= 1.0.0.0 && < 1.7
+                    , hedis >= 0.15.2 && < 0.16
+                    , safe >= 0.3 && < 0.4
+                    , safe-exceptions >= 0.1.7 && < 0.2
+                    , scientific >= 0.3.7.0 && < 0.4
+                    , stm >= 2.5.3 && < 3
+                    , text >= 1.2 && < 2.2
+                    , time >= 1.10 && < 1.15
+                    , units >= 2.4 && < 2.5
+                    , unix-time >= 0.4.11 && < 0.5
+
+
+                    , haskell-bee
+
+    -- Directories containing source files.
+    hs-source-dirs:   src
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      NumericUnderscores
+      OverloadedStrings
+      RecordWildCards
+
+test-suite redis-test-unit
+    -- Import common warning flags.
+    import:           warnings
+    
+    type:             exitcode-stdio-1.0
+    
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , tasty >= 1.5 && < 1.6
+                    , tasty-hunit >= 0.10 && < 0.11
+                    , tasty-quickcheck >= 0.10 && < 0.12
+                    , unix-time >= 0.4.11 && < 0.5
+                    
+                    , haskell-bee
+                    , haskell-bee-redis
+
+    -- Directories containing source files.
+    hs-source-dirs:   tests
+    main-is: unit-tests.hs
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      OverloadedStrings
+      RecordWildCards
+
+    ghc-options: -threaded
+
+test-suite redis-test-integration
+    -- Import common warning flags.
+    import:           warnings
+    
+    type:             exitcode-stdio-1.0
+
+    other-modules:  Test.Integration.Redis
+        
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , containers >= 0.6 && < 0.8
+                    , hedis >= 0.15.2 && < 0.16
+                    , hspec >= 2.11 && < 3
+                    , random-strings == 0.1.1.0
+                    , stm >= 2.5.3 && < 3
+                    , tasty >= 1.5 && < 1.6
+                    , tasty-hspec >= 1.2.0 && < 2
+                    , text >= 1.2 && < 2.2
+                    
+                    , haskell-bee
+                    , haskell-bee-tests
+                    , haskell-bee-redis
+
+    -- Directories containing source files.
+    hs-source-dirs:   tests
+    main-is: integration-tests.hs
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DeriveGeneric
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      OverloadedStrings
+      RecordWildCards
+
+    ghc-options: -threaded -fprof-auto

haskell-bee-redis/tests/Test/Integration/Redis.hs

+
+module Test.Integration.Redis
+ ( redisBrokerInitParams
+ , redisWorkerBrokerInitParams )
+where
+
+import Async.Worker.Broker.Redis qualified as BR
+import Async.Worker.Broker.Types qualified as BT
+import Async.Worker.Types (Job(..))
+import Data.Maybe (fromMaybe)
+import Database.Redis qualified as Redis
+import System.Environment (lookupEnv)
+import Test.Integration.Broker qualified as TIB
+import Test.Integration.Worker qualified as TIW
+
+
+-- | Redis connect info that is fetched from env
+getRedisEnvConnectInfo :: IO Redis.ConnectInfo
+getRedisEnvConnectInfo = do
+  redisHost <- lookupEnv "REDIS_HOST"
+  -- https://hackage.haskell.org/package/hedis-0.15.2/docs/Database-Redis.html#v:defaultConnectInfo
+  pure $ Redis.defaultConnectInfo { Redis.connectHost = fromMaybe "localhost" redisHost }
+
+
+redisBrokerInitParams :: IO (BT.BrokerInitParams BR.RedisBroker TIB.Message)
+redisBrokerInitParams = do
+  BR.RedisBrokerInitParams <$> getRedisEnvConnectInfo
+
+
+redisWorkerBrokerInitParams :: IO (BT.BrokerInitParams BR.RedisBroker (Job TIW.Message))
+redisWorkerBrokerInitParams = do
+  BR.RedisBrokerInitParams <$> getRedisEnvConnectInfo

haskell-bee-redis/tests/integration-tests.hs

+module Main where
+
+import Test.Integration.Broker (brokerTests)
+import Test.Integration.Redis (redisBrokerInitParams, redisWorkerBrokerInitParams)
+import Test.Integration.Worker (workerTests, multiWorkerTests)
+import Test.Tasty
+import Test.Tasty.Hspec
+
+
+
+main :: IO ()
+main = do
+  redisBInitParams <- redisBrokerInitParams
+  redisBrokerSpec <- testSpec "brokerTests" (brokerTests redisBInitParams)
+  redisWBInitParams <- redisWorkerBrokerInitParams
+  redisWorkerSpec <- testSpec "workerTests" (workerTests redisWBInitParams)
+  redisMultiWorkerSpec <- testSpec "multiWorkerTests" (multiWorkerTests redisWBInitParams 5)
+
+  defaultMain $ testGroup "Redis integration tests"
+    [
+      redisBrokerSpec
+    , redisWorkerSpec
+    , redisMultiWorkerSpec
+   ]
+

haskell-bee-redis/tests/unit-tests.hs

+{-# OPTIONS_GHC -Wno-orphans -Wno-missing-signatures #-}
+
+module Main where
+
+import Async.Worker.Broker.Redis qualified as R
+import Data.Aeson qualified as Aeson
+import Test.Tasty
+import Test.Tasty.QuickCheck as QC
+    
+
+main = defaultMain tests
+
+tests :: TestTree
+tests = testGroup "Tests" [propertyTests]
+
+
+
+propertyTests = testGroup "Property tests" [aesonPropTests]
+    
+aesonPropTests = testGroup "Aeson (de-)serialization property tests" $
+ [ aesonPropRedisTests ]
+
+
+instance QC.Arbitrary a => QC.Arbitrary (R.RedisWithMsgId a) where
+  arbitrary = do
+    rmidId <- arbitrary
+    rmida <- arbitrary
+    return $ R.RedisWithMsgId { rmida, rmidId }
+
+aesonPropRedisTests = testGroup "Aeson RedisWithMsgId (de-)serialization tests" $
+  [ QC.testProperty "Aeson.decode . Aeson.encode == id" $
+     \j ->
+       Aeson.decode (Aeson.encode (j :: R.RedisWithMsgId String)) == Just j
+  ]

haskell-bee-stm/haskell-bee-stm.cabal

+cabal-version:      3.4
+-- The cabal-version field refers to the version of the .cabal specification,
+-- and can be different from the cabal-install (the tool) version and the
+-- Cabal (the library) version you are using. As such, the Cabal (the library)
+-- version used must be equal or greater than the version stated in this field.
+-- Starting from the specification version 2.2, the cabal-version field must be
+-- the first thing in the cabal file.
+
+-- Initial package description 'haskell-pgmq' generated by
+-- 'cabal init'. For further documentation, see:
+--   http://haskell.org/cabal/users-guide/
+--
+-- The name of the package.
+name:               haskell-bee-stm
+
+-- The package version.
+-- See the Haskell package versioning policy (PVP) for standards
+-- guiding when and how versions should be incremented.
+-- https://pvp.haskell.org
+-- PVP summary:     +-+------- breaking API changes
+--                  | | +----- non-breaking API additions
+--                  | | | +--- code changes with no API change
+version:            0.1.0.0
+
+-- A short (one-line) description of the package.
+-- synopsis:
+
+-- A longer description of the package.
+-- description:
+
+-- URL for the project homepage or repository.
+homepage:           https://gitlab.iscpif.fr/gargantext/haskell-bee
+
+-- The license under which the package is released.
+license:            AGPL-3.0-or-later
+
+-- The file containing the license text.
+license-file:       ../LICENSE
+
+-- The package author(s).
+author:             Gargantext
+
+-- An email address to which users can send suggestions, bug reports, and patches.
+maintainer:         gargantext@iscpif.fr
+
+-- A copyright notice.
+-- copyright:
+category:           Concurrency
+build-type:         Simple
+
+-- Extra doc files to be distributed with the package, such as a CHANGELOG or a README.
+extra-doc-files:    CHANGELOG.md
+                  , README.md
+
+-- Extra source files to be distributed with the package, such as examples, or a tutorial module.
+-- extra-source-files:
+
+common warnings
+    ghc-options: -Wall
+
+library
+    -- Import common warning flags.
+    import:           warnings
+
+    -- Modules exported by the library.
+    exposed-modules:  Async.Worker.Broker.STM
+
+    -- Modules included in this library but not exported.
+    -- other-modules:
+
+    -- LANGUAGE extensions used by modules in this package.
+    -- other-extensions:
+
+    -- Other library packages from which modules are imported.
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , bytestring >= 0.11 && < 0.13
+                    , containers >= 0.6.7 && < 0.8
+                    , deepseq >= 1.0.0.0 && < 1.7
+                    , safe >= 0.3 && < 0.4
+                    , safe-exceptions >= 0.1.7 && < 0.2
+                    , scientific >= 0.3.7.0 && < 0.4
+                    , stm >= 2.5.3 && < 3
+                    , text >= 1.2 && < 2.2
+                    , time >= 1.10 && < 1.15
+                    , units >= 2.4 && < 2.5
+                    , unix-time >= 0.4.11 && < 0.5
+
+                    , haskell-bee
+
+    -- Directories containing source files.
+    hs-source-dirs:   src
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      NumericUnderscores
+      OverloadedStrings
+      RecordWildCards
+
+
+test-suite stm-test-unit
+    -- Import common warning flags.
+    import:           warnings
+    
+    type:             exitcode-stdio-1.0
+    
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , tasty >= 1.5 && < 1.6
+                    , tasty-hunit >= 0.10 && < 0.11
+                    , tasty-quickcheck >= 0.10 && < 0.12
+                    , unix-time >= 0.4.11 && < 0.5
+                    
+                    , haskell-bee
+                    , haskell-bee-stm
+
+    -- Directories containing source files.
+    hs-source-dirs:   tests
+    main-is: unit-tests.hs
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      OverloadedStrings
+      RecordWildCards
+
+    ghc-options: -threaded
+
+      
+test-suite stm-test-integration
+    -- Import common warning flags.
+    import:           warnings
+    
+    type:             exitcode-stdio-1.0
+
+    other-modules:  Test.Integration.STM
+        
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , containers >= 0.6 && < 0.8
+                    , hspec >= 2.11 && < 3
+                    , random-strings == 0.1.1.0
+                    , stm >= 2.5.3 && < 3
+                    , tasty >= 1.5 && < 1.6
+                    , tasty-hspec >= 1.2.0 && < 2
+                    , text >= 1.2 && < 2.2
+                    
+                    , haskell-bee
+                    , haskell-bee-tests
+                    , haskell-bee-stm
+
+    -- Directories containing source files.
+    hs-source-dirs:   tests
+    main-is: integration-tests.hs
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DeriveGeneric
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      OverloadedStrings
+      RecordWildCards
+
+    ghc-options: -threaded -fprof-auto

haskell-bee-stm/tests/Test/Integration/STM.hs

+module Test.Integration.STM
+ ( stmBrokerInitParams
+ , stmWorkerBrokerInitParams )
+where
+
+import Async.Worker.Broker.STM qualified as STMB
+import Async.Worker.Broker.Types qualified as BT
+import Async.Worker.Types (Job(..))
+import Control.Concurrent.STM.TVar (newTVarIO)
+import Data.Map.Strict qualified as Map
+import Test.Integration.Broker qualified as TIB
+import Test.Integration.Worker qualified as TIW
+
+
+stmBrokerInitParams :: IO (BT.BrokerInitParams STMB.STMBroker TIB.Message)
+stmBrokerInitParams = do
+  archiveMap <- newTVarIO Map.empty
+  stmMap <- newTVarIO Map.empty
+  pure $ STMB.STMBrokerInitParams { .. }
+
+
+stmWorkerBrokerInitParams :: IO (BT.BrokerInitParams STMB.STMBroker (Job TIW.Message))
+stmWorkerBrokerInitParams = do
+  archiveMap <- newTVarIO Map.empty
+  stmMap <- newTVarIO Map.empty
+  pure $  STMB.STMBrokerInitParams { .. }

haskell-bee-stm/tests/integration-tests.hs

+module Main where
+
+import Test.Integration.Broker (brokerTests)
+import Test.Integration.STM (stmBrokerInitParams, stmWorkerBrokerInitParams)
+import Test.Integration.Worker (workerTests, multiWorkerTests)
+import Test.Tasty
+import Test.Tasty.Hspec
+
+
+
+main :: IO ()
+main = do
+  stmBInitParams <- stmBrokerInitParams
+  stmBrokerSpec <- testSpec "brokerTests" (brokerTests stmBInitParams)
+  stmWBInitParams <- stmWorkerBrokerInitParams
+  stmWorkerSpec <- testSpec "workerTests" (workerTests stmWBInitParams)
+  stmMultiWorkerSpec <- testSpec "multiWorkerTests" (multiWorkerTests stmWBInitParams 5)
+
+  defaultMain $ testGroup "STM integration tests"
+    [
+      stmBrokerSpec
+    , stmWorkerSpec
+    , stmMultiWorkerSpec
+   ]
+

haskell-bee-stm/tests/unit-tests.hs

+{-# OPTIONS_GHC -Wno-orphans -Wno-missing-signatures #-}
+
+module Main where
+
+import Async.Worker.Broker.STM qualified as STMB
+import Data.Aeson qualified as Aeson
+import Data.UnixTime
+import Test.Tasty
+import Test.Tasty.QuickCheck as QC
+    
+
+main = defaultMain tests
+
+tests :: TestTree
+tests = testGroup "Tests" [propertyTests]
+
+
+
+propertyTests = testGroup "Property tests" [aesonPropTests]
+    
+aesonPropTests = testGroup "Aeson (de-)serialization property tests" $
+ [ aesonPropSTMBTests ]
+
+
+instance QC.Arbitrary a => QC.Arbitrary (STMB.STMWithMsgId a) where
+  arbitrary = do
+    stmidId <- arbitrary
+    stmida <- arbitrary
+    utSeconds <- arbitrary
+    utMicroSeconds <- arbitrary
+    let stmidInvisibleUntil = UnixTime { utSeconds, utMicroSeconds }
+    return $ STMB.STMWithMsgId { stmida, stmidId, stmidInvisibleUntil }
+
+aesonPropSTMBTests = testGroup "Aeson STMWithMsgId (de-)serialization tests" $
+  [ QC.testProperty "Aeson.decode . Aeson.encode == id" $
+     \j ->
+       Aeson.decode (Aeson.encode (j :: STMB.STMWithMsgId String)) == Just j
+  ]

haskell-bee-tests/haskell-bee-tests.cabal

+cabal-version:      3.4
+-- The cabal-version field refers to the version of the .cabal specification,
+-- and can be different from the cabal-install (the tool) version and the
+-- Cabal (the library) version you are using. As such, the Cabal (the library)
+-- version used must be equal or greater than the version stated in this field.
+-- Starting from the specification version 2.2, the cabal-version field must be
+-- the first thing in the cabal file.
+
+-- Initial package description 'haskell-bee' generated by
+-- 'cabal init'. For further documentation, see:
+--   http://haskell.org/cabal/users-guide/
+--
+-- The name of the package.
+name:               haskell-bee-tests
+
+-- The package version.
+-- See the Haskell package versioning policy (PVP) for standards
+-- guiding when and how versions should be incremented.
+-- https://pvp.haskell.org
+-- PVP summary:     +-+------- breaking API changes
+--                  | | +----- non-breaking API additions
+--                  | | | +--- code changes with no API change
+version:            0.1.0.0
+
+-- A short (one-line) description of the package.
+-- synopsis:
+
+-- A longer description of the package.
+-- description:
+
+-- URL for the project homepage or repository.
+homepage:           https://gitlab.iscpif.fr/gargantext/haskell-bee
+
+-- The license under which the package is released.
+license:            AGPL-3.0-or-later
+
+-- The file containing the license text.
+license-file:       ../LICENSE
+
+-- The package author(s).
+author:             Gargantext
+
+-- An email address to which users can send suggestions, bug reports, and patches.
+maintainer:         gargantext@iscpif.fr
+
+-- A copyright notice.
+-- copyright:
+category:           Concurrency
+build-type:         Simple
+
+-- Extra source files to be distributed with the package, such as examples, or a tutorial module.
+-- extra-source-files:
+
+common warnings
+    ghc-options: -Wall
+
+library
+    -- Import common warning flags.
+    import:           warnings
+    
+    type:             exitcode-stdio-1.0
+
+    exposed-modules:  Test.Integration.Broker
+                    , Test.Integration.Utils
+                    , Test.Integration.Worker
+        
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , containers >= 0.6 && < 0.8
+                    , hedis >= 0.15.2 && < 0.16
+                    , hspec >= 2.11 && < 3
+                    , postgresql-simple >= 0.6 && < 0.8
+                    , random-strings == 0.1.1.0
+                    , stm >= 2.5.3 && < 3
+                    , tasty >= 1.5 && < 1.6
+                    , tasty-hspec >= 1.2.0 && < 2
+                    , text >= 1.2 && < 2.2
+                    
+                    , haskell-bee
+
+    -- Directories containing source files.
+    hs-source-dirs:   tests
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DeriveGeneric
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      OverloadedStrings
+      RecordWildCards
+
+    ghc-options: -threaded -fprof-auto

tests/Test/Integration/Broker.hs → haskell-bee-tests/tests/Test/Integration/Broker.hs

@@ -7,24 +7,18 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Test.Integration.Broker
- ( brokerTests
- , pgmqBrokerInitParams
- , redisBrokerInitParams
- , stmBrokerInitParams )
+ ( Message(..)
+ 
+ , brokerTests )
 where
 
-import Async.Worker.Broker.PGMQ qualified as PGMQ
-import Async.Worker.Broker.Redis qualified as Redis
-import Async.Worker.Broker.STM qualified as STMB
 import Async.Worker.Broker.Types qualified as BT
-import Control.Concurrent.STM.TVar (newTVarIO)
 import Control.Exception (bracket)
 import Data.Aeson (ToJSON(..), FromJSON(..), withText)
-import Data.Map.Strict qualified as Map
 import Data.Maybe (isJust)
 import Data.Text qualified as T
 import Test.Hspec
-import Test.Integration.Utils (defaultPGMQVt, getPSQLEnvConnectInfo, getRedisEnvConnectInfo, randomQueueName, waitUntil)
+import Test.Integration.Utils (randomQueueName, waitUntil)
 import Test.RandomStrings (randomASCII, randomString, onlyAlphaNum)
 
 
@@ -158,19 +152,3 @@ brokerTests bInitParams =
       msgId <- BT.sendMessage broker queue (BT.toMessage msg)
       mMsg <- BT.getMessageById broker queue msgId
       (BT.toA . BT.getMessage <$> mMsg) `shouldBe` (Just msg)
-
-
-pgmqBrokerInitParams :: IO (BT.BrokerInitParams PGMQ.PGMQBroker Message)
-pgmqBrokerInitParams = do
-  conn <- getPSQLEnvConnectInfo
-  return $ PGMQ.PGMQBrokerInitParams conn defaultPGMQVt
-
-redisBrokerInitParams :: IO (BT.BrokerInitParams Redis.RedisBroker Message)
-redisBrokerInitParams = do
-  Redis.RedisBrokerInitParams <$> getRedisEnvConnectInfo
-
-stmBrokerInitParams :: IO (BT.BrokerInitParams STMB.STMBroker Message)
-stmBrokerInitParams = do
-  archiveMap <- newTVarIO Map.empty
-  stmMap <- newTVarIO Map.empty
-  pure $ STMB.STMBrokerInitParams { .. }

tests/Test/Integration/Utils.hs → haskell-bee-tests/tests/Test/Integration/Utils.hs

 {-# LANGUAGE OverloadedStrings #-}
 
 module Test.Integration.Utils
-  ( defaultPGMQVt
-  , getPSQLEnvConnectInfo
-  , getRedisEnvConnectInfo
-  , randomQueueName
+  ( randomQueueName
   , waitUntil
   , waitUntilTVarEq
   , waitUntilTVarPred
@@ -18,8 +15,6 @@ import Control.Concurrent.STM.TVar (TVar, readTVarIO)
 import Control.Monad (unless)
 import Data.Maybe (fromMaybe)
 import Data.String (fromString)
-import Database.PostgreSQL.Simple qualified as PSQL
-import Database.Redis qualified as Redis
 import System.Environment (lookupEnv)
 import System.Timeout qualified as Timeout
 import Test.Hspec (expectationFailure, shouldBe, shouldSatisfy, Expectation, HasCallStack)
@@ -32,40 +27,6 @@ newtype TimeoutMs = TimeoutMs Int
   deriving (Eq, Show, Num, Integral, Real, Enum, Ord)
 
 
--- | Visibility timeout is a very important parameter for PGMQ. It is
--- mainly used when reading a job: it specifies for how many seconds
--- this job should be invisible for other workers. We need more tests
--- and setting this correctly, preferably in accordance with
--- 'Job.timeout'. Issue is that at the broker level we don't know
--- anything about 'Job'...
---
--- The lower the value, the more probable that some other worker will
--- pick up the same job at about the same time (before broker marks it
--- as invisible).
-defaultPGMQVt :: Int
-defaultPGMQVt = 1
-
-
--- | PSQL connect info that is fetched from env
-getPSQLEnvConnectInfo :: IO PSQL.ConnectInfo
-getPSQLEnvConnectInfo = do
-  pgUser <- lookupEnv "POSTGRES_USER"
-  pgDb <- lookupEnv "POSTGRES_DB"
-  pgPass <- lookupEnv "POSTGRES_PASSWORD"
-  pgHost <- lookupEnv "POSTGRES_HOST"
-  -- https://hackage.haskell.org/package/postgresql-simple-0.7.0.0/docs/Database-PostgreSQL-Simple.html#t:ConnectInfo
-  pure $ PSQL.defaultConnectInfo { PSQL.connectUser = fromMaybe "postgres" pgUser
-                                 , PSQL.connectDatabase = fromMaybe "postgres" pgDb
-                                 , PSQL.connectHost = fromMaybe "localhost" pgHost
-                                 , PSQL.connectPassword = fromMaybe "postgres" pgPass }
-
--- | Redis connect info that is fetched from env
-getRedisEnvConnectInfo :: IO Redis.ConnectInfo
-getRedisEnvConnectInfo = do
-  redisHost <- lookupEnv "REDIS_HOST"
-  -- https://hackage.haskell.org/package/hedis-0.15.2/docs/Database-Redis.html#v:defaultConnectInfo
-  pure $ Redis.defaultConnectInfo { Redis.connectHost = fromMaybe "localhost" redisHost }
-
 -- | Given a queue prefix, add a random suffix to create a queue name
 randomQueueName :: B.Queue -> IO B.Queue
 randomQueueName prefix = do

tests/Test/Integration/Worker.hs → haskell-bee-tests/tests/Test/Integration/Worker.hs

@@ -11,17 +11,13 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Test.Integration.Worker
- ( workerTests
- , multiWorkerTests
- , pgmqWorkerBrokerInitParams
- , redisWorkerBrokerInitParams
- , stmWorkerBrokerInitParams )
+ ( Message(..)
+ 
+ , workerTests
+ , multiWorkerTests )
 where
 
 import Async.Worker (run, mkDefaultSendJob, mkDefaultSendJob', sendJob', errStrat, toStrat, resendOnKill, KillWorkerSafely(..))
-import Async.Worker.Broker.PGMQ qualified as PGMQ
-import Async.Worker.Broker.Redis qualified as Redis
-import Async.Worker.Broker.STM qualified as STMB
 import Async.Worker.Broker.Types qualified as BT
 import Async.Worker.Types
 import Control.Concurrent (forkIO, killThread, threadDelay, ThreadId, throwTo)
@@ -35,7 +31,7 @@ import Data.Maybe (fromJust, isJust)
 import Data.Set qualified as Set
 import GHC.Generics (Generic)
 import Test.Hspec
-import Test.Integration.Utils (defaultPGMQVt, getPSQLEnvConnectInfo, getRedisEnvConnectInfo, randomQueueName, waitUntil, waitUntilTVarEq, waitUntilTVarPred, waitUntil, waitUntilQueueEmpty)
+import Test.Integration.Utils (randomQueueName, waitUntil, waitUntilTVarEq, waitUntilTVarPred, waitUntil, waitUntilQueueEmpty)
 
 
 data TestEnv b =
@@ -553,19 +549,3 @@ second = 1000 * millisecond
 
 millisecond :: Int
 millisecond = 1000
-
-
-pgmqWorkerBrokerInitParams :: IO (BT.BrokerInitParams PGMQ.PGMQBroker (Job Message))
-pgmqWorkerBrokerInitParams = do
-  conn <- getPSQLEnvConnectInfo
-  return $ PGMQ.PGMQBrokerInitParams conn defaultPGMQVt
-
-redisWorkerBrokerInitParams :: IO (BT.BrokerInitParams Redis.RedisBroker (Job Message))
-redisWorkerBrokerInitParams = do
-  Redis.RedisBrokerInitParams <$> getRedisEnvConnectInfo
-
-stmWorkerBrokerInitParams :: IO (BT.BrokerInitParams STMB.STMBroker (Job Message))
-stmWorkerBrokerInitParams = do
-  archiveMap <- newTVarIO Map.empty
-  stmMap <- newTVarIO Map.empty
-  pure $  STMB.STMBrokerInitParams { .. }

haskell-bee/haskell-bee.cabal

+cabal-version:      3.4
+-- The cabal-version field refers to the version of the .cabal specification,
+-- and can be different from the cabal-install (the tool) version and the
+-- Cabal (the library) version you are using. As such, the Cabal (the library)
+-- version used must be equal or greater than the version stated in this field.
+-- Starting from the specification version 2.2, the cabal-version field must be
+-- the first thing in the cabal file.
+
+-- Initial package description 'haskell-bee' generated by
+-- 'cabal init'. For further documentation, see:
+--   http://haskell.org/cabal/users-guide/
+--
+-- The name of the package.
+name:               haskell-bee
+
+-- The package version.
+-- See the Haskell package versioning policy (PVP) for standards
+-- guiding when and how versions should be incremented.
+-- https://pvp.haskell.org
+-- PVP summary:     +-+------- breaking API changes
+--                  | | +----- non-breaking API additions
+--                  | | | +--- code changes with no API change
+version:            0.1.0.0
+
+-- A short (one-line) description of the package.
+-- synopsis:
+
+-- A longer description of the package.
+-- description:
+
+-- URL for the project homepage or repository.
+homepage:           https://gitlab.iscpif.fr/gargantext/haskell-bee
+
+-- The license under which the package is released.
+license:            AGPL-3.0-or-later
+
+-- The file containing the license text.
+license-file:       ../LICENSE
+
+-- The package author(s).
+author:             Gargantext
+
+-- An email address to which users can send suggestions, bug reports, and patches.
+maintainer:         gargantext@iscpif.fr
+
+-- A copyright notice.
+-- copyright:
+category:           Concurrency
+build-type:         Simple
+
+-- Extra doc files to be distributed with the package, such as a CHANGELOG or a README.
+extra-doc-files:    CHANGELOG.md
+                  , README.md
+
+-- Extra source files to be distributed with the package, such as examples, or a tutorial module.
+-- extra-source-files:
+
+common warnings
+    ghc-options: -Wall
+
+library
+    -- Import common warning flags.
+    import:           warnings
+
+    -- Modules exported by the library.
+    exposed-modules:  Async.Worker
+                    , Async.Worker.Broker
+                    , Async.Worker.Broker.Types
+                    , Async.Worker.Types
+
+    -- Modules included in this library but not exported.
+    -- other-modules:
+
+    -- LANGUAGE extensions used by modules in this package.
+    -- other-extensions:
+
+    -- Other library packages from which modules are imported.
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , bytestring >= 0.11 && < 0.13
+                    , containers >= 0.6.7 && < 0.8
+                    , deepseq >= 1.0.0.0 && < 1.7
+                    , mtl >= 2.2 && < 2.4
+                    , safe >= 0.3 && < 0.4
+                    , safe-exceptions >= 0.1.7 && < 0.2
+                    , scientific >= 0.3.7.0 && < 0.4
+                    , stm >= 2.5.3 && < 3
+                    , text >= 1.2 && < 2.2
+                    , time >= 1.10 && < 1.15
+                    , units >= 2.4 && < 2.5
+                    , unix-time >= 0.4.11 && < 0.5
+
+    -- Directories containing source files.
+    hs-source-dirs:   src
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      NumericUnderscores
+      OverloadedStrings
+      RecordWildCards
+
+test-suite test-unit
+    -- Import common warning flags.
+    import:           warnings
+    
+    type:             exitcode-stdio-1.0
+    
+    build-depends:    base ^>=4.17.2.0
+                    , aeson >= 2.1 && < 2.3
+                    , tasty >= 1.5 && < 1.6
+                    , tasty-hunit >= 0.10 && < 0.11
+                    , tasty-quickcheck >= 0.10 && < 0.12
+                    , unix-time >= 0.4.11 && < 0.5
+                    
+                    , haskell-bee
+
+    -- Directories containing source files.
+    hs-source-dirs:   tests
+    main-is: unit-tests.hs
+
+    -- Base language which the package is written in.
+    default-language: Haskell2010
+
+    default-extensions:
+      DuplicateRecordFields
+      GeneralizedNewtypeDeriving
+      ImportQualifiedPost
+      NamedFieldPuns
+      OverloadedStrings
+      RecordWildCards
+
+    ghc-options: -threaded
+

haskell-bee/src/Async/PGMQWorkerOld.hs

+{-|
+Module      : Database.PGMQ.Worker
+Description : PGMQ async worker implementation
+Copyright   : (c) Gargantext, 2024-Present
+License     : AGPL
+Maintainer  : gargantext@iscpif.fr
+Stability   : experimental
+Portability : POSIX
+
+NOTE: Although this module depends on 'Database.PGMQ.Simple', it does
+so only in a small way. In fact we could have a different mechanism
+for sending and reading tasks, as long as it can handle JSON
+serialization of our jobs with respective metadata.
+-}
+
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+    
+module Database.PGMQ.Worker
+  ( -- | reexports from Types
+    ArchiveStrategy(..)
+  , ErrorStrategy(..)
+  , TimeoutStrategy(..)
+  , JobMessage
+  , Job(..)
+  , getJob
+  , State
+
+  -- | worker functions
+  , newState
+  , run
+  , run'
+  , formatStr
+  , SendJob(..)
+  , mkDefaultSendJob
+  , sendJob )
+where
+
+import Control.Exception (SomeException, catch, fromException, throwIO)
+import Control.Monad (forever)
+import Data.Maybe (fromMaybe)
+import Database.PostgreSQL.Simple qualified as PSQL
+import Database.PGMQ.Simple qualified as PGMQ
+import Database.PGMQ.Types qualified as PGMQ
+import Database.PGMQ.Worker.Types
+import System.Timeout (timeout)
+
+
+-- | Helper function to define worker 'State'
+newState :: (PGMQ.SerializableMessage a)
+         => PSQL.ConnectInfo
+         -> String
+         -> PerformAction a
+         -> PGMQ.Queue
+         -> State a
+newState connectInfo name performAction queue = State { .. }
+  where
+    visibilityTimeout = 10
+    maxPollSeconds = 5
+    pollIntervalMs = 100
+    
+
+-- | Main function to start the worker
+run :: (PGMQ.SerializableMessage a)
+    => State a -> IO ()
+run state = do
+  conn <- PSQL.connect $ connectInfo state
+  run' state conn
+
+-- | Main function to start the worker (given a 'PSQL.Connection')
+run' :: forall a. (PGMQ.SerializableMessage a)
+     => State a -> PSQL.Connection -> IO ()
+run' state@(State { visibilityTimeout = stateVisibilityTimeout, .. }) conn = do
+  PGMQ.initialize conn
+  PGMQ.createQueue conn queue
+  -- The whole worker is just an infinite loop
+  forever loop
+  where
+    -- | Main looping function (describes a single loop iteration)
+    loop :: IO ()
+    loop = do
+      -- We catch any errors that could happen so that the worker runs
+      -- safely
+      catch (do
+        -- Read a message from queue and handle it.
+        PGMQ.readMessageWithPoll conn queue stateVisibilityTimeout maxPollSeconds pollIntervalMs >>=
+          handleMessage) handleLoopError
+
+    -- | Error handling function for the whole loop
+    handleLoopError :: SomeException -> IO ()
+    handleLoopError err = do
+      case fromException err of
+        -- TODO It can happen that the queue contains
+        -- ill-formatted messages. I don't yet know how to
+        -- handle this case - one would have to obtain message
+        -- id to properly delete that using pgmq.
+        Just (PSQL.ConversionFailed {}) -> do
+          putStrLn $ formatStr state $ show err
+        Just _ -> do
+          putStrLn $ formatStr state $ show err
+        Nothing -> do
+          putStrLn $ formatStr state $ "Exception: " <> show err
+
+    -- | Main job handling function. The message could not have
+    -- | arrived before read poll ended, hence the 'Maybe JobMessage'
+    -- | type.
+    handleMessage :: Maybe (JobMessage a) -> IO ()
+    handleMessage Nothing = return ()
+    handleMessage (Just msg@PGMQ.Message { message = Job { metadata = JobMetadata { .. } }, msgId }) = do
+      putStrLn $ formatStr state $ "handling message " <> show msgId <> " (AS: " <> show archiveStrategy <> ", ES: " <> show errorStrategy <> ")"
+      -- Immediately set visibility timeout of a job, when it's
+      -- specified in job's metadata (stateVisibilityTimeout is a
+      -- global property of the worker and might not reflect job's
+      -- timeout specs)
+      case visibilityTimeout of
+        Nothing -> pure ()
+        Just vt -> PGMQ.setMessageVt conn queue msgId vt
+      let vt = fromMaybe stateVisibilityTimeout visibilityTimeout
+    
+      let archiveHandler = do
+            case archiveStrategy of
+              ASDelete -> do
+                -- putStrLn $ formatStr state $ "deleting completed job " <> show msgId <> " (strategy: " <> show archiveStrategy <> ")"
+                PGMQ.deleteMessage conn queue msgId
+              ASArchive -> do
+                -- putStrLn $ formatStr state $ "archiving completed job " <> show msgId <> " (strategy: " <> show archiveStrategy <> ")"
+                PGMQ.archiveMessage conn queue msgId
+
+      -- Handle errors of 'performAction'.
+      let errorHandler :: SomeException -> IO ()
+          errorHandler err = do
+            case fromException err of
+              Just (JobTimeout { messageId }) -> do
+                let PGMQ.Message { readCt } = msg
+                putStrLn $ formatStr state $ "timeout occured: `" <> show timeoutStrategy <> " (readCt: " <> show readCt <> ", messageId = " <> show messageId <> ")"
+                
+                case timeoutStrategy of
+                  TSDelete -> PGMQ.deleteMessage conn queue messageId
+                  TSArchive -> PGMQ.archiveMessage conn queue messageId
+                  TSRepeat -> pure ()
+                  TSRepeatNElseArchive n -> do
+                    -- OK so this can be repeated at most 'n' times, compare 'readCt' with 'n'
+                    if readCt > n then
+                      PGMQ.archiveMessage conn queue messageId
+                    else
+                      pure ()
+                  TSRepeatNElseDelete n -> do
+                    -- OK so this can be repeated at most 'n' times, compare 'readCt' with 'n'
+                    if readCt > n then
+                      PGMQ.deleteMessage conn queue messageId
+                    else
+                      pure ()                      
+              _ -> do
+                putStrLn $ formatStr state $ "Error occured: `" <> show err
+                case errorStrategy of
+                  ESDelete -> PGMQ.deleteMessage conn queue msgId
+                  ESArchive -> PGMQ.deleteMessage conn queue msgId
+                  ESRepeat -> return ()
+
+      (do
+        mTimeout <- timeout (vt * microsecond) (performAction state msg)
+        case mTimeout of
+          Just _ -> archiveHandler
+          Nothing -> throwIO $ JobTimeout { messageId = msgId, vt = vt }
+        ) `catch` errorHandler
+
+        
+-- | Helper function to format a string with worker name (for logging)
+formatStr :: State a -> String -> String
+formatStr (State { name }) msg =
+  "[" <> name <> "] " <> msg
+
+microsecond :: Int
+microsecond = 10^6
+
+
+    
+
+-- | Wraps parameters for the 'sendJob' function
+data SendJob a =
+  SendJob { conn      :: PSQL.Connection
+          , queue     :: PGMQ.Queue
+          , msg       :: a
+          , delay     :: PGMQ.Delay
+          , archStrat :: ArchiveStrategy
+          , errStrat  :: ErrorStrategy
+          , toStrat   :: TimeoutStrategy
+          , vt        :: Maybe PGMQ.VisibilityTimeout }
+
+-- | Create a 'SendJob' data with some defaults
+mkDefaultSendJob :: PSQL.Connection
+               -> PGMQ.Queue
+               -> a
+               -> SendJob a
+mkDefaultSendJob conn queue msg =
+  SendJob { conn
+          , queue
+          , msg
+          , delay = 0
+          -- | remove finished jobs
+          , archStrat = ASDelete
+          -- | archive errored jobs (for inspection later)
+          , errStrat = ESArchive
+          -- | repeat timed out jobs
+          , toStrat = TSRepeat
+          , vt = Nothing }
+    
+    
+-- | Send given message as a worker job to pgmq. This wraps
+-- | 'PGMQ.sendMessage' with worker job metadata.
+sendJob :: (PGMQ.SerializableMessage a)
+        => SendJob a -> IO ()
+sendJob (SendJob { .. }) = do
+  let metadata = JobMetadata { archiveStrategy = archStrat
+                             , errorStrategy = errStrat
+                             , timeoutStrategy = toStrat
+                             , visibilityTimeout = vt }
+  PGMQ.sendMessage conn queue (Job { job = msg
+                                   , metadata = metadata }) delay

tests/unit-tests.hs → haskell-bee/tests/unit-tests.hs

@@ -2,11 +2,8 @@
 
 module Main where
 
-import Async.Worker.Broker.Redis qualified as R
-import Async.Worker.Broker.STM qualified as STMB
 import Async.Worker.Types qualified as WT
 import Data.Aeson qualified as Aeson
-import Data.UnixTime
 import Test.Tasty
 import Test.Tasty.QuickCheck as QC
     
@@ -22,9 +19,7 @@ propertyTests = testGroup "Property tests" [aesonPropTests]
     
 aesonPropTests = testGroup "Aeson (de-)serialization property tests" $
  [ aesonPropJobMetadataTests
- , aesonPropJobTests
- , aesonPropRedisTests
- , aesonPropSTMBTests ]
+ , aesonPropJobTests ]
 
 instance QC.Arbitrary WT.ArchiveStrategy where
   arbitrary = QC.elements [ WT.ASDelete, WT.ASArchive ]
@@ -66,34 +61,6 @@ aesonPropJobTests = testGroup "Aeson WT.Job (de-)serialization tests" $
         Aeson.decode (Aeson.encode (j :: WT.Job String)) == Just j
   ]
 
-
-instance QC.Arbitrary a => QC.Arbitrary (R.RedisWithMsgId a) where
-  arbitrary = do
-    rmidId <- arbitrary
-    rmida <- arbitrary
-    return $ R.RedisWithMsgId { rmida, rmidId }
-
-aesonPropRedisTests = testGroup "Aeson RedisWithMsgId (de-)serialization tests" $
-  [ QC.testProperty "Aeson.decode . Aeson.encode == id" $
-     \j ->
-       Aeson.decode (Aeson.encode (j :: R.RedisWithMsgId String)) == Just j
-  ]
-
-instance QC.Arbitrary a => QC.Arbitrary (STMB.STMWithMsgId a) where
-  arbitrary = do
-    stmidId <- arbitrary
-    stmida <- arbitrary
-    utSeconds <- arbitrary
-    utMicroSeconds <- arbitrary
-    let stmidInvisibleUntil = UnixTime { utSeconds, utMicroSeconds }
-    return $ STMB.STMWithMsgId { stmida, stmidId, stmidInvisibleUntil }
-
-aesonPropSTMBTests = testGroup "Aeson STMWithMsgId (de-)serialization tests" $
-  [ QC.testProperty "Aeson.decode . Aeson.encode == id" $
-     \j ->
-       Aeson.decode (Aeson.encode (j :: STMB.STMWithMsgId String)) == Just j
-  ]
-
     
 unitTests = testGroup "Unit tests" []
   -- [ testCase "List comparison (different length)" $

tests/integration-tests.hs

-module Main where
-
-import Test.Integration.Broker (brokerTests, pgmqBrokerInitParams, redisBrokerInitParams, stmBrokerInitParams)
-import Test.Integration.Worker (workerTests, multiWorkerTests, pgmqWorkerBrokerInitParams, redisWorkerBrokerInitParams, stmWorkerBrokerInitParams)
-import Test.Tasty
-import Test.Tasty.Hspec
-
-
-
-main :: IO ()
-main = do
-  pgmqBInitParams <- pgmqBrokerInitParams
-  pgmqBrokerSpec <- testSpec "brokerTests (pgmq)" (brokerTests pgmqBInitParams)
-  pgmqWBInitParams <- pgmqWorkerBrokerInitParams
-  pgmqWorkerSpec <- testSpec "workerTests (pgmq)" (workerTests pgmqWBInitParams)
-  pgmqMultiWorkerSpec <- testSpec "multiWorkerTests (pgmq)" (multiWorkerTests pgmqWBInitParams 5)
-
-  redisBInitParams <- redisBrokerInitParams
-  redisBrokerSpec <- testSpec "brokerTests (redis)" (brokerTests redisBInitParams)
-  redisWBInitParams <- redisWorkerBrokerInitParams
-  redisWorkerSpec <- testSpec "workerTests (redis)" (workerTests redisWBInitParams)
-  redisMultiWorkerSpec <- testSpec "multiWorkerTests (redis)" (multiWorkerTests redisWBInitParams 5)
-
-  stmBInitParams <- stmBrokerInitParams
-  stmBrokerSpec <- testSpec "brokerTests (stm)" (brokerTests stmBInitParams)
-  stmWBInitParams <- stmWorkerBrokerInitParams
-  stmWorkerSpec <- testSpec "workerTests (stm)" (workerTests stmWBInitParams)
-  stmMultiWorkerSpec <- testSpec "multiWorkerTests (stm)" (multiWorkerTests stmWBInitParams 5)
-
-  defaultMain $ testGroup "integration tests"
-    [
-      pgmqBrokerSpec
-    , pgmqWorkerSpec
-    , pgmqMultiWorkerSpec
-    
-    , redisBrokerSpec
-    , redisWorkerSpec
-    , redisMultiWorkerSpec
-
-    , stmBrokerSpec
-    , stmWorkerSpec
-    , stmMultiWorkerSpec 
-   ]
-