[FIX] conflicts

README.md

@@ -410,8 +410,6 @@ $ ALTER ROLE gargantua PASSWORD 'yourPasswordIn_gargantext-settings.toml'
 ```
 Maybe you need to change the port to 5433 for database connection in your gargantext.ini file.
 
-
-
 ## `haskell-language-server`
 
 If you want to use `haskell-language-server` for GHC 9.4.7, install it
@@ -420,3 +418,68 @@ with `ghcup`:
 ghcup compile hls --version 2.7.0.0 --ghc 9.4.7
 ```
 https://haskell-language-server.readthedocs.io/en/latest/installation.html
+
+
+# Async workers
+
+Async workers allow us to accept long-running jobs and execute them
+asynchronously. Workers can be spawned on multiple machines, which
+allows for scaling.
+
+To run the worker, follow these steps:
+- start a PostgreSQL DB, usually the one with Gargantext DB is enough
+- `"simple"` worker definition is in `gargantext-settings.toml`
+- run worker: `cabal v2-run gargantext-cli -- worker --name simple`
+
+When running the worker for the first time (or sending a job), best
+attempt is made to ensure the DB exists (if not, we will try to create
+it) and the `pgmq` schema is initialized. This allows for hassle-free
+maintenance and easier onboarding.
+
+The project that we base our worker is
+[haskell-bee](https://gitlab.iscpif.fr/gargantext/haskell-bee/). It's
+a more generic framework for managing asynchronous workers, supporting
+different brokers. Here, we decided to use `pgmq` because we already
+have PostgreSQL deployed.
+
+## Design
+
+Thanks to the fact that we already use `Servant.Jobs` (which executes
+the jobs in the gargantext-API process), we can migrate our jobs more
+easily to a different backend.
+
+All job definitions are in `G.A.A.EnvTypes` under `GargJob`
+datatype. However, because of a bit different design, the contsructors
+for this datatype are empty, without their respective arguments.
+
+Hence, I created `G.C.W.J.Types` with the `Job` datatype. Hopefully,
+in the future, we can replace `GargJob` with this datatype.
+
+If you want to add a new job, just add a new constructor to `Job`
+(with all the arguments this job needs), implement to/from JSON
+serialization (so we can send and read that job via the broker) and
+implement appropriate handler in `G.C.Worker` -> `performAction`.
+
+## No storage backend
+
+When you compare `haskell-bee` with the [celery
+project](https://docs.celeryq.dev/en/stable/), you can notice 2
+things:
+- both have configurable brokers
+- `haskell-bee` doesn't have a storage backend This is because, when
+analyzing our tasks, we actually don't store any **task** results
+anywhere. The progress can be reported immediately via notifications,
+same for errors. And when a task is executed (e.g. add to corpus with
+query), the garg db state is mutated.
+
+One could discuss if such a **task** storage backend is needed: if you
+want to debug some task, just run the worker and log its results. You
+can inspect later the errors in the log. On the other hand, if we had
+a production worker, which processes lots of tasks, it could be a pain
+to store the results of all of them, as we mostly don't care about the
+finished ones.
+
+Also, the `haskell-bee` framework allows to add custom hooks to the
+worker. In particular, search for `onJobError`/`onJobTimeout` in
+`Worker.State`. We could trigger some `IO` action on these hooks
+(logging, sending mail, firing rockets).

bin/gargantext-cli/CLI/Import.hs

@@ -33,7 +33,7 @@ import Gargantext.Database.Action.Flow.Types (FlowCmdM)
 import Gargantext.Database.Admin.Types.Node (CorpusId)
 import Gargantext.Database.Query.Tree.Root (MkCorpusUser(MkCorpusUserNormalCorpusName))
 import Gargantext.Prelude
-import Gargantext.Utils.Jobs.Monad ( MonadJobStatus, JobHandle )
+import Gargantext.Utils.Jobs.Monad ( MonadJobStatus(..) )
 import Options.Applicative
 import Prelude (String)
 import qualified Data.Text as T

bin/gargantext-cli/CLI/Ini.hs

@@ -28,6 +28,7 @@ import Gargantext.Core.Config.Ini.Ini qualified as Ini
 import Gargantext.Core.Config.Ini.Mail qualified as IniMail
 import Gargantext.Core.Config.Ini.NLP qualified as IniNLP
 import Gargantext.Core.Config.Types qualified as CTypes
+import Gargantext.Core.Config.Worker (WorkerSettings(..), WorkerDefinition(..))
 import Gargantext.Prelude
 import Options.Applicative
 import Servant.Client.Core (parseBaseUrl)
@@ -35,7 +36,7 @@ import Toml qualified
 
 
 iniCLI :: IniArgs -> IO ()
-iniCLI iniArgs = do
+iniCLI iniArgs@(IniArgs { dry_run }) = do
   let iniPath  = fromMaybe "gargantext.ini"           $ ini_path  iniArgs
   let tomlPath = fromMaybe "gargantext-settings.toml" $ toml_path iniArgs
   putStrLn $ "Reading configuration from file " <> iniPath <> "..."
@@ -44,8 +45,11 @@ iniCLI iniArgs = do
   iniNLP <- IniNLP.readConfig iniPath
   connInfo <- Ini.readDBConfig iniPath
   let c = convertConfigs ini iniMail iniNLP connInfo
-  T.writeFile tomlPath (show (Toml.encode c) :: Text)
-  putStrLn $ "Converted configuration into TOML and wrote it to file " <> tomlPath
+  if dry_run then
+    putStrLn (show (Toml.encode c) :: Text)
+  else do
+    T.writeFile tomlPath (show (Toml.encode c) :: Text)
+    putStrLn $ "Converted configuration into TOML and wrote it to file " <> tomlPath
 
 iniCmd :: HasCallStack => Mod CommandFields CLI
 iniCmd = command "ini" (info (helper <*> fmap CLISub iniParser)
@@ -54,7 +58,8 @@ iniCmd = command "ini" (info (helper <*> fmap CLISub iniParser)
 iniParser :: Parser CLICmd
 iniParser = fmap CCMD_ini $ IniArgs <$>
   (optional . strOption $ long "ini-path"  <> help "Path to the input ini file"   ) <*>
-  (optional . strOption $ long "toml-path" <> help "Path to the output .toml file")
+  (optional . strOption $ long "toml-path" <> help "Path to the output .toml file") <*>
+  (flag False True (long "dry-run" <> help "If set, will only output generated .toml file to stdout"))
 
 convertConfigs :: Ini.GargConfig -> IniMail.MailConfig -> IniNLP.NLPConfig -> PGS.ConnectInfo -> Config.GargConfig
 convertConfigs ini@(Ini.GargConfig { .. }) iniMail nlpConfig connInfo =
@@ -78,6 +83,9 @@ convertConfigs ini@(Ini.GargConfig { .. }) iniMail nlpConfig connInfo =
                                                    , _jc_js_id_timeout = _gc_js_id_timeout }
                     , _gc_apis = CTypes.APIsConfig { _ac_epo_api_url = _gc_epo_api_url
                                                    , _ac_scrapyd_url }
+                    , _gc_worker = WorkerSettings { _wsDefinitions = [ wd ]
+                                                  , _wsDefaultVisibilityTimeout = 1
+                                                  , _wsDatabase = connInfo { PGS.connectDatabase = "pgmq"} }
                     , _gc_log_level = LevelDebug
                     }
   where
@@ -85,6 +93,8 @@ convertConfigs ini@(Ini.GargConfig { .. }) iniMail nlpConfig connInfo =
       case parseBaseUrl "http://localhost:6800" of
         Nothing -> panicTrace "Cannot parse base url for scrapyd"
         Just b -> b
+    wd = WorkerDefinition { _wdName = "default"
+                          , _wdQueue = "default" }
 
 mkFrontendConfig :: Ini.GargConfig -> CTypes.FrontendConfig
 mkFrontendConfig (Ini.GargConfig { .. }) =

bin/gargantext-cli/CLI/Types.hs

@@ -48,6 +48,7 @@ data ImportArgs = ImportArgs
 data IniArgs = IniArgs
   { ini_path  :: !(Maybe FilePath)
   , toml_path :: !(Maybe FilePath)
+  , dry_run   :: !Bool
   } deriving (Show, Eq)
 
 data InitArgs = InitArgs
@@ -79,6 +80,12 @@ data CLIRoutes
   | CLIR_export FilePath
   deriving (Show, Eq)
 
+data WorkerArgs = WorkerArgs
+  { worker_toml       :: !SettingsFile
+  , worker_name       :: !Text
+  , worker_run_single :: !Bool
+  } deriving (Show, Eq)
+
 data CLICmd
   = CCMD_clean_csv_corpus
   | CCMD_filter_terms_and_cooc !CorpusFile !TermListFile !OutputFile
@@ -93,6 +100,7 @@ data CLICmd
   | CCMD_upgrade !UpgradeArgs
   | CCMD_golden_file_diff !GoldenFileDiffArgs
   | CCMD_routes !CLIRoutes
+  | CCMD_worker !WorkerArgs
   deriving (Show, Eq)
 
 data CLI =

bin/gargantext-cli/CLI/Worker.hs

+{-|
+Module      : Worker.hs
+Description : Gargantext Job Worker
+Copyright   : (c) CNRS, 2017-Present
+License     : AGPL + CECILL v3
+Maintainer  : team@gargantext.org
+Stability   : experimental
+Portability : POSIX
+
+ -}
+
+
+module CLI.Worker where
+
+import CLI.Types
+import CLI.Parsers
+import Data.List qualified as List (cycle, concat, take)
+import Data.Text qualified as T
+import Gargantext.Core.Config (hasConfig, _gc_worker)
+import Gargantext.Core.Config.Types (SettingsFile(..))
+import Gargantext.Core.Config.Worker (WorkerDefinition(..), WorkerSettings(..), findDefinitionByName)
+import Gargantext.Core.Worker (withPGMQWorker, withPGMQWorkerSingle)
+import Gargantext.Core.Worker.Env (withWorkerEnv)
+import Gargantext.Core.Worker.Jobs (sendJob)
+import Gargantext.Core.Worker.Jobs.Types (Job(Ping))
+import Gargantext.Prelude
+import Options.Applicative
+import Prelude qualified
+
+
+-- TODO Command to spawn all workers at once
+-- TODO Command to monitor queues
+-- TODO Support for KillWorkerSafely on Ctrl-C (so that the job in progress is moved back into the queue)
+
+workerCLI :: WorkerArgs -> IO ()
+workerCLI (WorkerArgs { .. }) = do
+
+  let ___ = putStrLn ((List.concat
+                       $ List.take 72
+                       $ List.cycle ["_"]) :: Prelude.String)
+
+  ___
+  putStrLn ("GarganText worker" :: Text)
+  putStrLn ("worker_name: " <> worker_name)
+  putStrLn ("worker toml: " <> _SettingsFile worker_toml)
+  ___
+
+  withWorkerEnv worker_toml $ \env -> do
+    let gc = env ^. hasConfig
+    let ws = _gc_worker gc
+    case findDefinitionByName ws worker_name of
+      Nothing -> do
+        let workerNames = _wdName <$> (_wsDefinitions ws)
+        let availableWorkers = T.intercalate ", " workerNames
+        putStrLn ("Worker definition not found! Available workers: " <> availableWorkers)
+      Just wd -> do
+        putStrLn ("Starting worker '" <> worker_name <> "'")
+        putStrLn ("Worker settings: " <> show ws :: Text)
+        if worker_run_single then
+          withPGMQWorkerSingle env wd $ \a _state -> do
+            wait a
+        else
+          withPGMQWorker env wd $ \a _state -> do
+            runReaderT (sendJob Ping) env
+            wait a
+
+
+workerCmd :: HasCallStack => Mod CommandFields CLI
+workerCmd = command "worker" (info (helper <*> fmap CLISub worker_p) (progDesc "Gargantext worker."))
+
+worker_p :: Parser CLICmd
+worker_p = fmap CCMD_worker $ WorkerArgs
+  <$> settings_p
+  <*> strOption ( long "name"
+                <> metavar "STRING"
+                <> help "Worker name, as defined in the .toml file" )
+  <*> flag False True ( long "run-single"
+                     <> help "Whether to loop or run a single job from queue" )
+

bin/gargantext-cli/Main.hs

@@ -33,6 +33,7 @@ import CLI.Phylo (phyloCLI, phyloCmd)
 import CLI.Phylo.Profile (phyloProfileCLI, phyloProfileCmd)
 import CLI.Server.Routes (routesCLI, routesCmd)
 import CLI.Upgrade (upgradeCLI, upgradeCmd)
+import CLI.Worker (workerCLI, workerCmd)
 
 runCLI :: CLI -> IO ()
 runCLI = \case
@@ -62,6 +63,8 @@ runCLI = \case
     -> fileDiffCLI args
   CLISub (CCMD_routes args)
     -> routesCLI args
+  CLISub (CCMD_worker args)
+    -> workerCLI args
 
 
 main :: IO ()
@@ -85,5 +88,6 @@ allOptions = subparser (
                  phyloProfileCmd <>
                  upgradeCmd <>
                  fileDiffCmd <>
-                 routesCmd
+                 routesCmd <>
+                 workerCmd
                  )

bin/update-project-dependencies

@@ -18,8 +18,8 @@ fi
 # with the `sha256sum` result calculated on the `cabal.project` and
 # `cabal.project.freeze`. This ensures the files stay deterministic so that CI
 # cache can kick in.
-expected_cabal_project_hash="3afb11e01938b74ae8419caa160180d8f8628a67315a2d689c3a42a76463071e"
-expected_cabal_project_freeze_hash="de1726d350936da5f5e15140e3be29bb4f44757c5702defe995c2386f1b4a741"
+expected_cabal_project_hash="eb28225cf0ebf07cc233223d3bbed085ea6ed19e05912c06ecda92a89f8132cb"
+expected_cabal_project_freeze_hash="30dd1cf2cb2015351dd0576391d22b187443b1935c2be23599b821ad1ab95f23"
 
 
 cabal --store-dir=$STORE_DIR v2-build --dry-run

cabal.project

@@ -190,14 +190,24 @@ source-repository-package
 
 source-repository-package
     type: git
-    location: https://github.com/glguy/toml-parser
-    tag: toml-parser-2.0.1.0
+    location: https://gitlab.iscpif.fr/gargantext/haskell-pgmq
+    tag: 0591a643d8ba1776af4fac56c1e4ff5fc3e98bb3
+
+source-repository-package
+    type: git
+    location: https://gitlab.iscpif.fr/gargantext/haskell-bee
+    tag: d783198e1b7ca8a61e5332965db468da3faef796
 
 source-repository-package
     type: git
     location: https://gitlab.iscpif.fr/gargantext/haskell-throttle
     tag: 02f5ed9ee2d6cce45161addf945b88bc6adf9059
 
+source-repository-package
+    type: git
+    location: https://github.com/glguy/toml-parser
+    tag: toml-parser-2.0.1.0
+
         
 allow-newer:
              accelerate-arithmetic:accelerate
@@ -231,6 +241,8 @@ allow-newer:
            , stemmer:base
 allow-older: aeson:hashable
            , crawlerHAL:servant-client
+           , haskell-bee:postgresql-libpq
+           , haskell-bee:stm
            , haskell-throttle:time
            , hsparql:rdf4h
 

cabal.project.freeze

 active-repositories: hackage.haskell.org:merge
-constraints: any.Cabal ==3.8.1.0,
+constraints: any.Boolean ==0.2.4,
+             any.Cabal ==3.8.1.0,
              any.Cabal-syntax ==3.8.1.0,
              any.Glob ==0.10.2,
              any.HTTP ==4000.4.1,
@@ -11,6 +12,7 @@ constraints: any.Cabal ==3.8.1.0,
              any.MissingH ==1.6.0.1,
              MissingH +network--ge-3_0_0,
              any.MonadRandom ==0.6,
+             any.NumInstances ==1.4,
              any.OneTuple ==0.4.2,
              any.Only ==0.1,
              any.QuickCheck ==2.14.3,
@@ -87,6 +89,7 @@ constraints: any.Cabal ==3.8.1.0,
              any.bytestring ==0.11.5.3,
              any.bytestring-builder ==0.10.8.2.0,
              bytestring-builder +bytestring_has_builder,
+             any.bytestring-lexing ==0.5.0.14,
              any.bzlib-conduit ==0.3.0.3,
              any.c2hs ==0.28.8,
              c2hs +base3 -regression,
@@ -180,6 +183,7 @@ constraints: any.Cabal ==3.8.1.0,
              entropy -donotgetentropy,
              any.epo-api-client ==0.1.0.0,
              any.erf ==2.0.0.0,
+             any.errors ==2.3.0,
              any.exceptions ==0.10.5,
              any.extra ==1.7.16,
              any.fail ==4.9.0.0,
@@ -220,12 +224,16 @@ constraints: any.Cabal ==3.8.1.0,
              hashable +integer-gmp -random-initial-seed,
              any.hashtables ==1.3.1,
              hashtables -bounds-checking -debug -detailed-profiling -portable -sse42 +unsafe-tricks,
+             any.haskell-bee ==0.1.0.0,
              any.haskell-igraph ==0.10.4,
              any.haskell-lexer ==1.1.1,
+             any.haskell-pgmq ==0.1.0.0,
              any.haskell-src-exts ==1.23.1,
              any.haskell-src-meta ==0.8.14,
              any.haskell-throttle ==0.1.0.0,
              any.hedgehog ==1.5,
+             any.hedis ==0.15.2,
+             hedis -dev,
              any.hgal ==2.0.0.3,
              any.hlcm ==0.2.2,
              any.hmatrix ==0.20.2,
@@ -298,6 +306,8 @@ constraints: any.Cabal ==3.8.1.0,
              any.libyaml-clib ==0.2.5,
              any.lifted-async ==0.10.2.5,
              any.lifted-base ==0.2.3.12,
+             any.linear ==1.23,
+             linear -herbie +template-haskell,
              any.list-t ==1.0.5.7,
              any.llvm-hs ==12.0.0,
              llvm-hs -debug -llvm-with-rtti +shared-llvm,
@@ -341,6 +351,7 @@ constraints: any.Cabal ==3.8.1.0,
              any.mtl ==2.2.2,
              any.mtl-compat ==0.2.2,
              mtl-compat -two-point-one -two-point-two,
+             any.multimap ==1.2.1,
              any.mwc-random ==0.15.1.0,
              mwc-random -benchpapi,
              any.nanomsg-haskell ==0.2.4,
@@ -351,6 +362,7 @@ constraints: any.Cabal ==3.8.1.0,
              any.network-control ==0.0.2,
              any.network-info ==0.2.1,
              any.network-uri ==2.6.4.2,
+             any.newtype-generics ==0.6.2,
              any.old-locale ==1.0.0.7,
              any.old-time ==1.1.0.4,
              any.opaleye ==0.9.7.0,
@@ -434,6 +446,7 @@ constraints: any.Cabal ==3.8.1.0,
              any.rts ==1.0.2,
              any.safe ==0.3.21,
              any.safe-exceptions ==0.1.7.4,
+             any.scanner ==0.3.1,
              any.scientific ==0.3.7.0,
              scientific -bytestring-builder -integer-simple,
              any.securemem ==0.1.10,
@@ -555,6 +568,8 @@ constraints: any.Cabal ==3.8.1.0,
              any.unicode-collation ==0.1.3.6,
              unicode-collation -doctests -executable,
              any.unique ==0.0.1,
+             any.units ==2.4.1.5,
+             any.units-parser ==0.1.1.5,
              any.unix ==2.7.3,
              any.unix-compat ==0.7.2,
              any.unix-time ==0.4.15,
@@ -578,6 +593,7 @@ constraints: any.Cabal ==3.8.1.0,
              any.vector-algorithms ==0.9.0.2,
              vector-algorithms +bench +boundschecks -internalchecks -llvm +properties -unsafechecks,
              any.vector-binary-instances ==0.2.5.2,
+             any.vector-space ==0.16,
              any.vector-th-unbox ==0.2.2,
              any.void ==0.7.3,
              void -safe,

gargantext-settings.toml_toModify

@@ -138,3 +138,25 @@ dispatcher = { bind = "tcp://*:5561", connect = "tcp://localhost:5561" }
 # - johnsnows:// (for https:// JohnSnow)
 EN = "spacy://localhost:8000"
 FR = "spacy://localhost:8001"
+
+[worker]
+
+# After this number of seconds, the job will be available again.
+# You can set timeout for each job individually and this is the
+# preferred method over using defaultVt.
+default_visibility_timeout = 1
+
+# if you leave the same credentials as in [database] section above,
+#  workers will try to set up the `gargantext_pgmq` database
+#  automatically
+[worker.database]
+host = "127.0.0.1"
+port = 5432
+name = "gargantext_pgmq"
+user = "gargantua"
+pass = PASSWORD_TO_CHANGE
+
+[[worker.definitions]]
+name = "default"
+queue = "default"
+

gargantext.cabal

@@ -172,6 +172,7 @@ library
       Gargantext.Core.Config.NLP
       Gargantext.Core.Config.Types
       Gargantext.Core.Config.Utils
+      Gargantext.Core.Config.Worker
       Gargantext.Core.Mail.Types
       Gargantext.Core.Methods.Similarities
       Gargantext.Core.Methods.Similarities.Conditional
@@ -239,6 +240,11 @@ library
       Gargantext.Core.Viz.Phylo.PhyloTools
       Gargantext.Core.Viz.Phylo.SynchronicClustering
       Gargantext.Core.Viz.Types
+      Gargantext.Core.Worker
+      Gargantext.Core.Worker.Broker
+      Gargantext.Core.Worker.Env
+      Gargantext.Core.Worker.Jobs
+      Gargantext.Core.Worker.Jobs.Types
       Gargantext.Database.Action.Flow
       Gargantext.Database.Action.Flow.Types
       Gargantext.Database.Action.Metrics.TFICF
@@ -518,7 +524,9 @@ library
     , gargantext-graph >=0.1.0.0
     , gargantext-prelude
     , graphviz ^>= 2999.20.1.0
+    , haskell-bee
     , haskell-igraph ^>= 0.10.4
+    , haskell-pgmq >= 0.1.0.0 && < 0.2
     , haskell-throttle
     , hlcm ^>= 0.2.2
     , hsinfomap ^>= 0.1
@@ -585,6 +593,7 @@ library
     , servant-swagger-ui-core >= 0.3.5
     , servant-websockets >= 2.0.0 && < 2.1
     , servant-xml-conduit ^>= 0.1.0.4
+    , shelly >= 1.9 && < 2
     , singletons ^>= 3.0.2
     , singletons-th >= 3.1 && < 3.2
     , smtp-mail >= 0.3.0.0
@@ -643,6 +652,7 @@ executable gargantext-cli
       CLI.Server.Routes
       CLI.Types
       CLI.Upgrade
+      CLI.Worker
       Paths_gargantext
   hs-source-dirs:
       bin/gargantext-cli
@@ -658,6 +668,7 @@ executable gargantext-cli
     , extra ^>= 1.7.9
     , gargantext
     , gargantext-prelude
+    , haskell-bee
     , ini ^>= 0.4.1
     , lens >= 5.2.2 && < 5.3
     , monad-logger ^>= 0.3.36
@@ -810,6 +821,7 @@ test-suite garg-test-tasty
       Test.Core.Text.Examples
       Test.Core.Text.Flow
       Test.Core.Utils
+      Test.Core.Worker
       Test.Database.Operations
       Test.Database.Operations.DocumentSearch
       Test.Database.Operations.NodeStory

src/Gargantext/API/Admin/Auth.hs

@@ -40,11 +40,12 @@ module Gargantext.API.Admin.Auth
   , withNamedAccess
 
   , ForgotPasswordAsyncParams
+
+  , forgotUserPassword
   )
   where
 
 import Control.Lens (view, (#))
-import Data.Text qualified as Text
 import Data.Text.Lazy.Encoding qualified as LE
 import Data.UUID (UUID, fromText, toText)
 import Data.UUID.V4 (nextRandom)
@@ -58,7 +59,8 @@ import Gargantext.Core.Config (HasJWTSettings(..))
 import Gargantext.Core.Mail (MailModel(..), mail)
 import Gargantext.Core.Mail.Types (mailSettings)
 import Gargantext.Core.Types.Individu (User(..), Username, GargPassword(..))
-import Gargantext.Database.Action.Flow.Types (FlowCmdM)
+import Gargantext.Core.Worker.Jobs qualified as Jobs
+import Gargantext.Core.Worker.Jobs.Types qualified as Jobs
 import Gargantext.Database.Action.User.New (guessUserName)
 import Gargantext.Database.Admin.Types.Node (NodeId(..))
 import Gargantext.Database.Admin.Types.Node (UserId)
@@ -70,7 +72,7 @@ import Gargantext.Database.Schema.Node (NodePoly(_node_id))
 import Gargantext.Prelude hiding (Handler, reverse, to)
 import Gargantext.Prelude.Crypto.Auth qualified as Auth
 import Gargantext.Prelude.Crypto.Pass.User (gargPass)
-import Gargantext.Utils.Jobs (serveJobsAPI, MonadJobStatus(..))
+import Gargantext.Utils.Jobs (serveJobsAPI)
 import Servant
 import Servant.API.Generic ()
 import Servant.Auth.Server
@@ -240,14 +242,7 @@ forgotPassword = Named.ForgotPasswordAPI
 
 forgotPasswordPost :: (CmdCommon env)
      => ForgotPasswordRequest -> Cmd' env err ForgotPasswordResponse
-forgotPasswordPost (ForgotPasswordRequest email) = do
-  us <- getUsersWithEmail (Text.toLower email)
-  case us of
-    [u] -> forgotUserPassword u
-    _ -> pure ()
-
-  -- NOTE Sending anything else here could leak information about
-  -- users' emails
+forgotPasswordPost (ForgotPasswordRequest _email) = do
   pure $ ForgotPasswordResponse "ok"
 
 forgotPasswordGet :: (CmdCommon env, HasServerError err)
@@ -327,19 +322,5 @@ generateForgotPasswordUUID = do
 -- malicious users emails of our users in the db
 forgotPasswordAsync :: Named.ForgotPasswordAsyncAPI (AsServerT (GargM Env BackendInternalError))
 forgotPasswordAsync = Named.ForgotPasswordAsyncAPI $ AsyncJobs $
-  serveJobsAPI ForgotPasswordJob $ \jHandle p -> forgotPasswordAsync' p jHandle
-
-forgotPasswordAsync' :: (FlowCmdM env err m, MonadJobStatus m)
-  => ForgotPasswordAsyncParams
-  -> JobHandle m
-  -> m ()
-forgotPasswordAsync' (ForgotPasswordAsyncParams { email }) jobHandle = do
-
-  markStarted 2 jobHandle
-  markProgress 1 jobHandle
-
-  -- printDebug "[forgotPasswordAsync'] email" email
-
-  _ <- forgotPasswordPost $ ForgotPasswordRequest { _fpReq_email = email }
-
-  markComplete jobHandle
+  serveJobsAPI ForgotPasswordJob $ \_jHandle p -> do
+    Jobs.sendJob $ Jobs.ForgotPasswordAsync { Jobs._fpa_args = p }

src/Gargantext/API/Admin/Auth/Types.hs

@@ -118,23 +118,23 @@ type Email = Text
 type Password = Text
 
 data ForgotPasswordRequest = ForgotPasswordRequest { _fpReq_email :: Email }
-  deriving (Generic )
+  deriving (Generic)
 instance ToSchema ForgotPasswordRequest where
   declareNamedSchema = genericDeclareNamedSchema (unPrefixSwagger "_fpReq_")
 
 data ForgotPasswordResponse = ForgotPasswordResponse { _fpRes_status :: Text }
-  deriving (Generic )
+  deriving (Generic)
 instance ToSchema ForgotPasswordResponse where
   declareNamedSchema = genericDeclareNamedSchema (unPrefixSwagger "_fpRes_")
 
 data ForgotPasswordGet = ForgotPasswordGet {_fpGet_password :: Password}
-  deriving (Generic )
+  deriving (Generic)
 instance ToSchema ForgotPasswordGet where
   declareNamedSchema = genericDeclareNamedSchema (unPrefixSwagger "_fpGet_")
 
 newtype ForgotPasswordAsyncParams =
   ForgotPasswordAsyncParams { email :: Text }
-  deriving (Generic, Show)
+  deriving (Generic, Show, Eq)
 instance FromJSON ForgotPasswordAsyncParams where
   parseJSON = genericParseJSON defaultOptions
 instance ToJSON ForgotPasswordAsyncParams where

src/Gargantext/API/Admin/EnvTypes.hs

@@ -6,6 +6,7 @@
 
 module Gargantext.API.Admin.EnvTypes (
     GargJob(..)
+  , parseGargJob
   , Env(..)
   , Mode(..)
   , modeToLoggingLevels
@@ -28,9 +29,12 @@ module Gargantext.API.Admin.EnvTypes (
   , ConcreteJobHandle -- opaque
   ) where
 
-import Control.Lens hiding (Level, (:<))
+import Control.Lens hiding (Level, (:<), (.=))
 import Control.Monad.Except
 import Control.Monad.Reader
+import Data.Aeson qualified as Aeson
+import Data.Aeson ((.:), (.=), object, withObject)
+import Data.Aeson.Types (prependFailure, typeMismatch)
 import Data.List ((\\))
 import Data.Pool (Pool)
 import Data.Sequence (ViewL(..), viewl)
@@ -99,24 +103,69 @@ instance HasLogger (GargM Env BackendInternalError) where
 
 
 data GargJob
-  = TableNgramsJob
-  | ForgotPasswordJob
-  | UpdateNgramsListJobJSON
-  | UpdateNgramsListJobTSV
+  = AddAnnuaireFormJob
   | AddContactJob
+  | AddCorpusFileJob
+  | AddCorpusFormJob
+  | AddCorpusQueryJob
   | AddFileJob
   | DocumentFromWriteNodeJob
-  | UpdateNodeJob
-  | UploadFrameCalcJob
-  | UploadDocumentJob
+  | ForgotPasswordJob
   | NewNodeJob
-  | AddCorpusQueryJob
-  | AddCorpusFormJob
-  | AddCorpusFileJob
-  | AddAnnuaireFormJob
   | RecomputeGraphJob
+  | TableNgramsJob
+  | UpdateNgramsListJobJSON
+  | UpdateNgramsListJobTSV
+  | UpdateNodeJob
+  | UploadDocumentJob
+  | UploadFrameCalcJob
   deriving (Show, Eq, Ord, Enum, Bounded)
 
+parseGargJob :: Text -> Maybe GargJob
+parseGargJob s = case s of
+  "addannuaireform" -> Just AddAnnuaireFormJob
+  "addcontact" -> Just AddContactJob
+  "addcorpusfile" -> Just AddCorpusFileJob
+  "addcorpusform" -> Just AddCorpusFormJob
+  "addcorpusquery" -> Just AddCorpusQueryJob
+  "addfile" -> Just AddFileJob
+  "documentfromwritenode" -> Just DocumentFromWriteNodeJob
+  "forgotpassword" -> Just ForgotPasswordJob
+  "newnode" -> Just NewNodeJob
+  "recomputegraph" -> Just RecomputeGraphJob
+  "tablengrams" -> Just TableNgramsJob
+  "updatedocument" -> Just UploadDocumentJob
+  "updateframecalc" -> Just UploadFrameCalcJob
+  "updatengramslistjson" -> Just UpdateNgramsListJobJSON
+  "updatengramslisttsv" -> Just UpdateNgramsListJobTSV
+  "updatenode" -> Just UpdateNodeJob
+  _ -> Nothing
+
+instance FromJSON GargJob where
+  parseJSON = withObject "GargJob" $ \o -> do
+    type_ <- o .: "type"
+    case parseGargJob type_ of
+      Just gj -> return gj
+      Nothing -> prependFailure "parsing garg job type failed, " (typeMismatch "type" $ Aeson.String type_)
+instance ToJSON GargJob where
+  toJSON AddAnnuaireFormJob = object [ ("type" .= ("addannuaireform" :: Text))]
+  toJSON AddContactJob = object [ ("type" .= ("addcontact" :: Text))]
+  toJSON AddCorpusFileJob = object [ ("type" .= ("addcorpusfile" :: Text))]
+  toJSON AddCorpusFormJob = object [ ("type" .= ("addcorpusform" :: Text))]
+  toJSON AddCorpusQueryJob = object [ ("type" .= ("addcorpusquery" :: Text))]
+  toJSON AddFileJob = object [ ("type" .= ("addfile" :: Text))]
+  toJSON DocumentFromWriteNodeJob = object [ ("type" .= ("documentfromwritenode" :: Text))]
+  toJSON ForgotPasswordJob = object [ ("type" .= ("forgotpassword" :: Text))]
+  toJSON NewNodeJob = object [ ("type" .= ("newnode" :: Text))]
+  toJSON RecomputeGraphJob = object [ ("type" .= ("recomputegraph" :: Text))]
+  toJSON TableNgramsJob = object [ ("type" .= ("tablengrams" :: Text))]
+  toJSON UploadDocumentJob = object [ ("type" .= ("updatedocument" :: Text))]
+  toJSON UploadFrameCalcJob = object [ ("type" .= ("updateframecalc" :: Text))]
+  toJSON UpdateNgramsListJobJSON = object [ ("type" .= ("updatengramslistjson" :: Text))]
+  toJSON UpdateNgramsListJobTSV = object [ ("type" .= ("updatengramslisttsv" :: Text))]
+  toJSON UpdateNodeJob = object [ ("type" .= ("updatenode" :: Text))]
+
+
 -- Do /not/ treat the data types of this type as strict, because it's convenient
 -- to be able to partially initialise things like an 'Env' during tests, without
 -- having to specify /everything/. This means that when we /construct/ an 'Env',

src/Gargantext/API/Admin/Settings/TOML.hs

+{-# LANGUAGE TemplateHaskell #-}
+module Gargantext.API.Admin.Settings.TOML where
+
+import Control.Lens hiding ((.=))
+import Data.Text qualified as T
+import Gargantext.API.Admin.Settings.CORS
+import Gargantext.API.Admin.Settings.MicroServices
+import Gargantext.Core.Worker.TOML
+import Gargantext.Prelude (panicTrace)
+import Gargantext.System.Logging
+import Prelude
+import Toml
+import Servant.Client.Core.BaseUrl
+
+-- | Compatibility bridge until we fix #304 (move to Toml)
+data GargTomlSettings = GargTomlSettings
+  { _gargCorsSettings          :: !CORSSettings
+  , _gargMicroServicesSettings :: !MicroServicesSettings
+  , _gargWorkerSettings        :: !WorkerSettings
+  }
+
+makeLenses ''GargTomlSettings
+
+settingsCodec :: TomlCodec GargTomlSettings
+settingsCodec = GargTomlSettings
+  <$> (Toml.table corsSettingsCodec "cors" .= _gargCorsSettings)
+  <*> (Toml.table  microServicesSettingsCodec "microservices.proxy" .= _gargMicroServicesSettings)
+  <*> (Toml.table workerSettingsCodec "worker" .= _gargWorkerSettings)
+
+-- | Extends the 'allowed-origins' in the CORSettings with the URLs embellished
+-- with the proxy port.
+addProxyToAllowedOrigins :: GargTomlSettings -> GargTomlSettings
+addProxyToAllowedOrigins stgs =
+  stgs & over gargCorsSettings (addProxies $ stgs ^. gargMicroServicesSettings . msProxyPort)
+  where
+    addProxies :: Int -> CORSSettings -> CORSSettings
+    addProxies port cors =
+      let origins = _corsAllowedOrigins cors
+          mkUrl (CORSOrigin bh) = CORSOrigin $ bh { baseUrlPort = port }
+      in cors { _corsAllowedOrigins = origins <> Prelude.map mkUrl origins }
+
+-- | Loads the 'CORSSettings' from the 'toml' file.
+loadGargTomlSettings :: FilePath -> IO GargTomlSettings
+loadGargTomlSettings tomlFile = do
+  tomlRes <- Toml.decodeFileEither settingsCodec tomlFile
+  case tomlRes of
+      Left errs      -> do
+        withLogger () $ \ioLogger -> do
+          logMsg ioLogger ERROR $ T.unpack $ "Error, gargantext-settings.toml parsing failed: " <> Toml.prettyTomlDecodeErrors errs
+          panicTrace "Please fix the errors in your gargantext-settings.toml file."
+      Right settings0 -> case settings0 ^. gargCorsSettings . corsUseOriginsForHosts of
+        True  -> pure $ addProxyToAllowedOrigins $
+                   settings0 & over (gargCorsSettings . corsAllowedHosts)
+                     (\_ -> (settings0 ^. gargCorsSettings . corsAllowedOrigins))
+        False -> pure $ addProxyToAllowedOrigins settings0

src/Gargantext/API/GraphQL.hs

@@ -13,10 +13,10 @@ Portability : POSIX
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DuplicateRecordFields #-}  -- permit duplications for field names in multiple constructors
 {-# LANGUAGE KindSignatures #-}  -- for use of Endpoint (name :: Symbol)
-{-# LANGUAGE PartialTypeSignatures #-}  -- to automatically use suggested type hole signatures during compilation
-{-# LANGUAGE TypeOperators #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE PartialTypeSignatures #-}  -- to automatically use suggested type hole signatures during compilation
 {-# LANGUAGE TypeApplications #-}
+{-# LANGUAGE TypeOperators #-}
 
 module Gargantext.API.GraphQL where
 

src/Gargantext/API/Node/New.hs

@@ -27,31 +27,33 @@ import Gargantext.API.Errors.Types
 import Gargantext.API.Node.New.Types
 import Gargantext.API.Prelude
 import Gargantext.API.Routes.Named.Node qualified as Named
-import Gargantext.Core.Notifications.CentralExchange.Types qualified as CE
-import Gargantext.Database.Action.Flow.Types
+import Gargantext.Core.AsyncUpdates.CentralExchange.Types qualified as CE
+import Gargantext.Core.Mail.Types (HasMail)
+import Gargantext.Core.NLP (HasNLPServer)
+import Gargantext.Core.Worker.Jobs qualified as Jobs
+import Gargantext.Core.Worker.Jobs.Types qualified as Jobs
 import Gargantext.Database.Action.Node
 import Gargantext.Database.Admin.Types.Node
-import Gargantext.Database.Prelude (DBCmd')
+import Gargantext.Database.Prelude (CmdM, DBCmd')
 import Gargantext.Database.Query.Table.Node.Error (HasNodeError(..))
 import Gargantext.Prelude
-import Gargantext.Utils.Jobs (serveJobsAPI, MonadJobStatus(..))
+import Gargantext.Utils.Jobs (serveJobsAPI)
 import Servant.Server.Generic (AsServerT)
 
 ------------------------------------------------------------------------
-postNode :: (HasNodeError err, CE.HasCentralExchangeNotification env)
+-- postNode :: (CmdM env err m, HasNodeError err, HasSettings env)
+postNode :: ( HasMail env
+            , HasNodeError err
+            , HasNLPServer env
+            , CE.HasCentralExchangeNotification env)
          => AuthenticatedUser
          -- ^ The logged-in user
          -> NodeId
          -> PostNode
+         -- -> m [NodeId]
          -> DBCmd' env err [NodeId]
-postNode authenticatedUser pId (PostNode nodeName nt) = do
-  let userId = authenticatedUser ^. auth_user_id
-  nodeIds <- mkNodeWithParent nt (Just pId) userId nodeName
-
-  -- mapM_ (CE.ce_notify . CE.UpdateTreeFirstLevel) nodeIds
-  CE.ce_notify $ CE.UpdateTreeFirstLevel pId
-  
-  return nodeIds
+postNode authenticatedUser nId pn = do
+  postNode' authenticatedUser nId pn
 
 postNodeAsyncAPI
   :: AuthenticatedUser
@@ -60,29 +62,57 @@ postNodeAsyncAPI
   -- ^ The target node
   -> Named.PostNodeAsyncAPI (AsServerT (GargM Env BackendInternalError))
 postNodeAsyncAPI authenticatedUser nId = Named.PostNodeAsyncAPI $ AsyncJobs $
-  serveJobsAPI NewNodeJob $ \jHandle p -> postNodeAsync authenticatedUser nId p jHandle
+  serveJobsAPI NewNodeJob $ \_jHandle p -> do
+    Jobs.sendJob $ Jobs.NewNodeAsync { Jobs._nna_node_id = nId
+                                     , Jobs._nna_authenticatedUser = authenticatedUser
+                                     , Jobs._nna_postNode = p }
+    -- postNodeAsync authenticatedUser nId p jHandle
 
 ------------------------------------------------------------------------
-postNodeAsync :: (FlowCmdM env err m, MonadJobStatus m, CE.HasCentralExchangeNotification env)
+-- postNode' :: (CmdM env err m, HasSettings env, HasNodeError err)
+--           => AuthenticatedUser
+--           -- ^ The logged-in user
+--           -> NodeId
+--           -> PostNode
+--           -> m [NodeId]
+-- postNode' authenticatedUser pId (PostNode nodeName nt) = do
+postNode' :: ( CmdM env err m
+             , HasNodeError err
+             , HasMail env
+             , CE.HasCentralExchangeNotification env)
     => AuthenticatedUser
     -- ^ The logged in user
     -> NodeId
     -> PostNode
-    -> JobHandle m
-    -> m ()
-postNodeAsync authenticatedUser nId (PostNode nodeName tn) jobHandle = do
-
-  -- printDebug "postNodeAsync" nId
-  markStarted 3 jobHandle
-  markProgress 1 jobHandle
-
-  -- _ <- threadDelay 1000
-  markProgress 1 jobHandle
+    -> m [NodeId]
+postNode' authenticatedUser nId (PostNode nodeName tn) = do
 
   let userId = authenticatedUser ^. auth_user_id
-  _nodeIds <- mkNodeWithParent tn (Just nId) userId nodeName
+  nodeIds <- mkNodeWithParent tn (Just nId) userId nodeName
 
   -- mapM_ (CE.ce_notify . CE.UpdateTreeFirstLevel) nodeIds
   CE.ce_notify $ CE.UpdateTreeFirstLevel nId
 
-  markComplete jobHandle
+  return nodeIds
+
+
+-- postNodeAsync :: (FlowCmdM env err m, MonadJobStatus m, HasSettings env)
+--     => AuthenticatedUser
+--     -- ^ The logged in user
+--     -> NodeId
+--     -> PostNode
+--     -> JobHandle m
+--     -> m ()
+-- postNodeAsync authenticatedUser nId (PostNode nodeName tn) jobHandle = do
+
+--   -- printDebug "postNodeAsync" nId
+--   markStarted 3 jobHandle
+--   markProgress 1 jobHandle
+
+--   -- _ <- threadDelay 1000
+--   markProgress 1 jobHandle
+
+--   let userId = authenticatedUser ^. auth_user_id
+--   _ <- mkNodeWithParent tn (Just nId) userId nodeName
+
+--   markComplete jobHandle

src/Gargantext/API/Node/New/Types.hs

+{-|
+Module      : Gargantext.API.Node.New.Types
+Description : 
+Copyright   : (c) CNRS, 2024-Present
+License     : AGPL + CECILL v3
+Maintainer  : team@gargantext.org
+Stability   : experimental
+Portability : POSIX
+
+-}
 
 module Gargantext.API.Node.New.Types (
   PostNode(..)
@@ -5,16 +15,16 @@ module Gargantext.API.Node.New.Types (
 
 import Data.Aeson
 import Data.Swagger
-import Data.Text (Text)
 import GHC.Generics
 import Gargantext.Core.Types (NodeType (..))
+import Gargantext.Prelude
 import Test.QuickCheck
 import Web.FormUrlEncoded
 
 ------------------------------------------------------------------------
 data PostNode = PostNode { pn_name     :: Text
-                         , pn_typename :: NodeType}
-  deriving (Generic)
+                         , pn_typename :: NodeType }
+  deriving (Generic, Eq, Show)
 ------------------------------------------------------------------------
 -- TODO unPrefix "pn_" FromJSON, ToJSON, ToSchema, adapt frontend.
 instance FromJSON  PostNode

src/Gargantext/API/Routes.hs

@@ -10,31 +10,26 @@ Portability : POSIX
 -}
 
 {-# LANGUAGE ConstraintKinds      #-}
-{-# LANGUAGE KindSignatures       #-}
 {-# LANGUAGE ScopedTypeVariables  #-}
-{-# LANGUAGE TypeApplications     #-}
 {-# LANGUAGE TypeFamilies         #-}
-{-# LANGUAGE TypeOperators        #-}
 {-# LANGUAGE UndecidableInstances #-}
 
 module Gargantext.API.Routes
       where
 
-import Control.Lens (view)
 import Data.Validity
 import Gargantext.API.Admin.EnvTypes (Env, GargJob(..))
 import Gargantext.API.Admin.Orchestrator.Types (AsyncJobs(..))
 import Gargantext.API.Errors.Types
 import Gargantext.API.Node.Corpus.Annuaire qualified as Annuaire
-import Gargantext.API.Node.Corpus.New qualified as New
 import Gargantext.API.Prelude
 import Gargantext.API.Routes.Named.Annuaire qualified as Named
 import Gargantext.API.Routes.Named.Corpus qualified as Named
-import Gargantext.Core.Config (gc_jobs, HasConfig(..))
-import Gargantext.Core.Config.Types (jc_max_docs_scrapers)
 import Gargantext.Core.Types.Individu (User(..))
+import Gargantext.Core.Worker.Jobs qualified as Jobs
+import Gargantext.Core.Worker.Jobs.Types qualified as Jobs
 import Gargantext.Prelude
-import Gargantext.Utils.Jobs (serveJobsAPI, MonadJobStatus(..))
+import Gargantext.Utils.Jobs (serveJobsAPI)
 import Servant
 import Servant.Auth.Swagger ()
 import Servant.Server.Generic (AsServerT)
@@ -53,9 +48,12 @@ waitAPI n = do
 
 addCorpusWithQuery :: User -> Named.AddWithQuery (AsServerT (GargM Env BackendInternalError))
 addCorpusWithQuery user = Named.AddWithQuery $ \cid -> AsyncJobs $
-  serveJobsAPI AddCorpusQueryJob $ \jHandle q -> do
-    limit <- view $ hasConfig . gc_jobs . jc_max_docs_scrapers
-    New.addToCorpusWithQuery user cid q (Just $ fromIntegral limit) jHandle
+  serveJobsAPI AddCorpusQueryJob $ \_jHandle q -> do
+    -- limit <- view $ hasConfig . gc_jobs . jc_max_docs_scrapers
+    -- New.addToCorpusWithQuery user cid q (Just $ fromIntegral limit) jHandle
+    Jobs.sendJob $ Jobs.AddCorpusWithQuery { Jobs._acq_args = q
+                                           , Jobs._acq_user = user
+                                           , Jobs._acq_cid = cid }
       {- let log' x = do
         printDebug "addToCorpusWithQuery" x
         liftBase $ log x
@@ -63,11 +61,15 @@ addCorpusWithQuery user = Named.AddWithQuery $ \cid -> AsyncJobs $
 
 addCorpusWithForm :: User -> Named.AddWithForm (AsServerT (GargM Env BackendInternalError))
 addCorpusWithForm user = Named.AddWithForm $ \cid -> AsyncJobs $
-  serveJobsAPI AddCorpusFormJob $ \jHandle i -> do
+  serveJobsAPI AddCorpusFormJob $ \_jHandle i -> do
     -- /NOTE(adinapoli)/ Track the initial steps outside 'addToCorpusWithForm', because it's
     -- called in a few places, and the job status might be different between invocations.
-    markStarted 3 jHandle
-    New.addToCorpusWithForm user cid i jHandle
+    -- markStarted 3 jHandle
+    -- New.addToCorpusWithForm user cid i jHandle
+    Jobs.sendJob $ Jobs.AddCorpusFormAsync { Jobs._acf_args = i
+                                           , Jobs._acf_user = user
+                                           , Jobs._acf_cid = cid }
+
 
 --addCorpusWithFile :: User -> ServerT Named.AddWithFile (GargM Env BackendInternalError)
 --addCorpusWithFile user cid =

src/Gargantext/Core/Config.hs

@@ -28,6 +28,7 @@ module Gargantext.Core.Config (
   , gc_jobs
   , gc_secrets
   , gc_apis
+  , gc_worker
   , gc_log_level
 
   , mkProxyUrl
@@ -42,6 +43,7 @@ import Data.Text as T
 import Database.PostgreSQL.Simple qualified as PSQL
 import Gargantext.Core.Config.Mail (MailConfig)
 import Gargantext.Core.Config.NLP (NLPConfig)
+import Gargantext.Core.Config.Worker (WorkerSettings)
 import Gargantext.Core.Config.Types
 import Gargantext.Prelude
 import Servant.Auth.Server (JWTSettings)
@@ -65,6 +67,7 @@ data GargConfig = GargConfig { _gc_datafilepath         :: ~FilePath
                              , _gc_jobs                 :: ~JobsConfig
                              , _gc_secrets              :: ~SecretsConfig
                              , _gc_apis                 :: ~APIsConfig
+                             , _gc_worker               :: ~WorkerSettings
                              , _gc_log_level            :: ~LogLevel
                              }
   deriving (Generic, Show)
@@ -83,6 +86,7 @@ instance FromValue GargConfig where
     _gc_jobs <- reqKey "jobs"
     _gc_apis <- reqKey "apis"
     _gc_notifications_config <- reqKey "notifications"
+    _gc_worker <- reqKey "worker"
     let _gc_log_level = LevelDebug
     return $ GargConfig { _gc_datafilepath
                         , _gc_jobs
@@ -94,6 +98,7 @@ instance FromValue GargConfig where
                         , _gc_notifications_config
                         , _gc_frames
                         , _gc_secrets
+                        , _gc_worker
                         , _gc_log_level }
 instance ToValue GargConfig where
   toValue = defaultTableToValue
@@ -109,6 +114,7 @@ instance ToTable GargConfig where
           , "mail" .= _gc_mail_config
           , "notifications" .= _gc_notifications_config
           , "nlp" .= _gc_nlp_config
+          , "worker" .= _gc_worker
           ]
 
 

src/Gargantext/Core/Config/Worker.hs

+{-|
+Module      : Gargantext.Core.Config.Worker
+Description : Worker TOML file config
+Copyright   : (c) CNRS, 2024
+License     : AGPL + CECILL v3
+Maintainer  : team@gargantext.org
+Stability   : experimental
+Portability : POSIX
+
+Although Async.Worker.Broker supports various Broker types, in
+Gargantext we will only use PGMQ. This makes for easier config,
+simpler design. Also, the DevOps stuff is simpler (providing multiple
+brokers at the same time could lead to complexities in analyzing
+what's going on).
+
+If need arises, we could switch to a different broker by rewriting its
+initialization. At the same time, sending and executing jobs should be
+broker-agnostic.
+
+-}
+
+
+module Gargantext.Core.Config.Worker where
+
+import Async.Worker.Broker.Types qualified as Broker
+import Database.PGMQ.Types qualified as PGMQ
+import Database.PostgreSQL.Simple qualified as PGS
+import Gargantext.Core.Config.Types (unTOMLConnectInfo, TOMLConnectInfo(..))
+import Gargantext.Prelude
+import Toml.Schema
+
+
+type WorkerName = Text
+
+
+data WorkerSettings =
+  WorkerSettings {
+      _wsDatabase     :: !PGS.ConnectInfo
+      -- After this number of seconds, the job will be available again.
+      
+      -- You can set timeout for each job individually and this is the
+      -- preferred method over using defaultVt.
+    , _wsDefaultVisibilityTimeout :: PGMQ.VisibilityTimeout
+    , _wsDefinitions  :: ![WorkerDefinition]
+  } deriving (Show, Eq)
+instance FromValue WorkerSettings where
+  fromValue = parseTableFromValue $ do
+    dbConfig <- reqKey "database"
+    _wsDefinitions <- reqKey "definitions"
+    _wsDefaultVisibilityTimeout <- reqKey "default_visibility_timeout"
+    return $ WorkerSettings { _wsDatabase = unTOMLConnectInfo dbConfig
+                            , _wsDefinitions
+                            , _wsDefaultVisibilityTimeout }
+instance ToValue WorkerSettings where
+  toValue = defaultTableToValue
+instance ToTable WorkerSettings where
+  toTable (WorkerSettings { .. }) =
+    table [ "database" .= TOMLConnectInfo _wsDatabase
+          , "default_visibility_timeout" .= _wsDefaultVisibilityTimeout
+          , "definitions" .= _wsDefinitions ]
+
+data WorkerDefinition =
+  WorkerDefinition {
+    _wdName   :: !WorkerName
+  , _wdQueue  :: !Broker.Queue
+  } deriving (Show, Eq)
+instance FromValue WorkerDefinition where
+  fromValue = parseTableFromValue $ do
+    _wdName <- reqKey "name"
+    queue <- reqKey "queue"
+    return $ WorkerDefinition { _wdQueue = Broker.Queue queue, .. }
+instance ToValue WorkerDefinition where
+  toValue = defaultTableToValue
+instance ToTable WorkerDefinition where
+  toTable (WorkerDefinition { .. }) =
+    table [ "name" .= _wdName
+          , "queue" .= Broker._Queue _wdQueue ]
+
+findDefinitionByName :: WorkerSettings -> WorkerName -> Maybe WorkerDefinition
+findDefinitionByName (WorkerSettings { _wsDefinitions }) workerName =
+  head $ filter (\wd -> _wdName wd == workerName) _wsDefinitions
+
+-- wdToRedisBrokerInitParams :: WorkerDefinition -> Maybe BRedis.RedisBrokerInitParams
+-- wdToRedisBrokerInitParams wd = BRedis.RedisBrokerInitParams <$> (wdToRedisConnectInfo wd)
+
+

src/Gargantext/Core/Types/Individu.hs

@@ -31,6 +31,7 @@ import Prelude qualified
 data User = UserDBId UserId | UserName Text | RootId NodeId
   deriving (Show, Eq, Generic)
 
+instance FromJSON User
 instance ToJSON User
 
 renderUser :: User -> T.Text

src/Gargantext/Core/Worker.hs

+{-|
+Module      : Gargantext.Core.Worker
+Description : Asynchronous worker logic
+Copyright   : (c) CNRS, 2024
+License     : AGPL + CECILL v3
+Maintainer  : team@gargantext.org
+Stability   : experimental
+Portability : POSIX
+
+-}
+
+{-# LANGUAGE UndecidableInstances #-}
+{-# OPTIONS_GHC -Wno-orphans      #-}  -- orphan HasNodeError IOException
+
+
+module Gargantext.Core.Worker where
+
+
+import Async.Worker.Broker.PGMQ (PGMQBroker)
+import Async.Worker.Broker.Types (BrokerMessage, toA, getMessage)
+import Async.Worker qualified as Worker
+import Async.Worker.Types qualified as Worker
+import Async.Worker.Types (HasWorkerBroker)
+import Data.Text qualified as T
+import Gargantext.API.Admin.Auth (forgotUserPassword)
+import Gargantext.API.Admin.Auth.Types (ForgotPasswordAsyncParams(..))
+import Gargantext.API.Node.Corpus.New (addToCorpusWithForm, addToCorpusWithQuery)
+import Gargantext.API.Node.New (postNode')
+import Gargantext.Core.Config (hasConfig, gc_jobs)
+import Gargantext.Core.Config.Types (jc_max_docs_scrapers)
+import Gargantext.Core.Config.Worker (WorkerDefinition(..))
+import Gargantext.Core.Worker.Broker (initBrokerWithDBCreate)
+import Gargantext.Core.Worker.Env
+import Gargantext.Core.Worker.Jobs.Types (Job(..))
+import Gargantext.Database.Query.Table.User (getUsersWithEmail)
+import Gargantext.Prelude
+import Gargantext.Utils.Jobs.Monad ( MonadJobStatus(noJobHandle) )
+
+
+
+-- | Spawn a worker with PGMQ broker
+-- TODO:
+--  - reduce size of DB pool
+--  - progress report via notifications
+--  - I think there is no point to save job result, as usually there is none (we have side-effects only)
+--  - replace Servant.Job to use workers instead of garg API threads
+withPGMQWorker :: (HasWorkerBroker PGMQBroker Job)
+                => WorkerEnv
+                -> WorkerDefinition
+                -> (Async () -> Worker.State PGMQBroker Job -> IO ())
+                -> IO ()
+withPGMQWorker env (WorkerDefinition { .. }) cb = do
+  let gargConfig = env ^. hasConfig
+  broker <- initBrokerWithDBCreate gargConfig
+
+  let state' = Worker.State { broker
+                            , queueName = _wdQueue
+                            , name = T.unpack _wdName
+                            , performAction = performAction env
+                            , onMessageReceived = Nothing
+                            , onJobFinish = Nothing
+                            , onJobTimeout = Nothing
+                            , onJobError = Nothing }
+
+  withAsync (Worker.run state') (\a -> cb a state')
+
+withPGMQWorkerSingle :: (HasWorkerBroker PGMQBroker Job)
+                     => WorkerEnv
+                     -> WorkerDefinition
+                     -> (Async () -> Worker.State PGMQBroker Job -> IO ())
+                     -> IO ()
+withPGMQWorkerSingle env (WorkerDefinition { .. }) cb = do
+  let gargConfig = env ^. hasConfig
+  broker <- initBrokerWithDBCreate gargConfig
+
+  let state' = Worker.State { broker
+                            , queueName = _wdQueue
+                            , name = T.unpack _wdName
+                            , performAction = performAction env
+                            , onMessageReceived = Nothing
+                            , onJobFinish = Nothing
+                            , onJobTimeout = Nothing
+                            , onJobError = Nothing }
+
+  withAsync (Worker.runSingle state') (\a -> cb a state')
+
+
+-- | How the worker should process jobs
+performAction :: (HasWorkerBroker b Job)
+              => WorkerEnv
+              -> Worker.State b Job
+              -> BrokerMessage b (Worker.Job Job)
+              -> IO ()
+performAction env _state bm = do
+  let job' = toA $ getMessage bm
+  case Worker.job job' of
+    Ping -> putStrLn ("ping" :: Text)
+    AddCorpusFormAsync { .. } -> runWorkerMonad env $ do
+      liftBase $ putStrLn ("add corpus form" :: Text)
+      addToCorpusWithForm _acf_user _acf_cid _acf_args (noJobHandle (Proxy :: Proxy WorkerMonad))
+    AddCorpusWithQuery { .. } -> runWorkerMonad env $ do
+      liftBase $ putStrLn ("add corpus with query" :: Text)
+      let limit = Just $ fromIntegral $ env ^. hasConfig . gc_jobs . jc_max_docs_scrapers
+      addToCorpusWithQuery _acq_user _acq_cid _acq_args limit (noJobHandle (Proxy :: Proxy WorkerMonad))
+    ForgotPasswordAsync { _fpa_args = ForgotPasswordAsyncParams { email } } -> runWorkerMonad env $ do
+      liftBase $ putStrLn ("forgot password: " <> email)
+      us <- getUsersWithEmail (T.toLower email)
+      case us of
+        [u] -> forgotUserPassword u
+        _ -> pure ()
+    NewNodeAsync { .. } -> runWorkerMonad env $ do
+      liftBase $ putStrLn ("new node async " :: Text)
+      void $ postNode' _nna_authenticatedUser _nna_node_id _nna_postNode
+      return ()
+    GargJob { _gj_garg_job } -> putStrLn ("Garg job: " <> show _gj_garg_job <> " (handling of this job is still not implemented!)" :: Text)

src/Gargantext/Core/Worker/Broker.hs

+{-# LANGUAGE TupleSections #-}
+
+module Gargantext.Core.Worker.Broker
+  ( initBrokerWithDBCreate )
+where
+
+import Async.Worker.Broker.PGMQ (PGMQBroker, BrokerInitParams(PGMQBrokerInitParams))
+import Async.Worker.Broker.Types (Broker, initBroker)
+import Async.Worker.Types qualified as WorkerT
+import Data.Text qualified as T
+import Data.Text.Encoding qualified as TE
+import Database.PostgreSQL.Simple qualified as PSQL
+import Gargantext.Core.Config (GargConfig(..), gc_worker)
+import Gargantext.Core.Config.Worker (WorkerSettings(..))
+import Gargantext.Core.Worker.Jobs.Types (Job(..))
+import Gargantext.Prelude
+import Shelly qualified as SH
+
+
+
+-- | Create DB if not exists, then run 'initBroker' (which, in
+-- particular, creates the pgmq extension, if needed)
+initBrokerWithDBCreate :: (WorkerT.HasWorkerBroker PGMQBroker Job)
+                       => GargConfig
+                       -> IO (Broker PGMQBroker (WorkerT.Job Job))
+initBrokerWithDBCreate gc@(GargConfig { _gc_database_config }) = do
+  -- By using gargantext db credentials, we create pgmq db (if needed)
+  let WorkerSettings { .. }  = gc ^. gc_worker
+  let psqlDB = TE.decodeUtf8 $ PSQL.postgreSQLConnectionString _gc_database_config
+  -- For the \gexec trick, see:
+  -- https://stackoverflow.com/questions/18389124/simulate-create-database-if-not-exists-for-postgresql
+  (_res, _ec) <- SH.shelly $ SH.silently $ SH.escaping False $ do
+    let sql = "\"SELECT 'CREATE DATABASE " <> (T.pack $ PSQL.connectDatabase _wsDatabase) <> "' WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '" <> (T.pack $ PSQL.connectDatabase _wsDatabase) <> "')\\gexec\""
+    result <- SH.run "echo" [sql, "|", "psql", "-d", "\"" <> psqlDB <> "\""]
+    (result,) <$> SH.lastExitCode
+
+  initBroker $ PGMQBrokerInitParams _wsDatabase _wsDefaultVisibilityTimeout

src/Gargantext/Core/Worker/Env.hs

@@ -24,8 +24,7 @@ import Data.Text qualified as T
 import Database.PostgreSQL.Simple (Connection)
 import Gargantext.API.Admin.EnvTypes (ConcreteJobHandle, GargJob, Mode(Dev), modeToLoggingLevels)
 import Gargantext.API.Admin.Orchestrator.Types (JobLog, noJobLog)
-import Gargantext.API.Admin.Settings ( devSettings, newPool )
-import Gargantext.API.Admin.Types (HasSettings(..), Settings(..))
+import Gargantext.API.Admin.Settings ( newPool )
 import Gargantext.API.Prelude (GargM)
 import Gargantext.Core.Notifications.CentralExchange qualified as CE
 import Gargantext.Core.Notifications.CentralExchange.Types qualified as CET
@@ -49,8 +48,7 @@ import System.Log.FastLogger qualified as FL
 
 
 data WorkerEnv = WorkerEnv
-  { _w_env_settings  :: !Settings
-  , _w_env_config    :: !GargConfig
+  { _w_env_config    :: !GargConfig
   , _w_env_logger    :: !(Logger (GargM WorkerEnv IOException))
   , _w_env_pool      :: !(Pool Connection)
   , _w_env_nodeStory :: !NodeStoryEnv
@@ -70,12 +68,10 @@ withWorkerEnv settingsFile k = withLoggerHoisted Dev $ \logger -> do
       --nodeStory_env <- fromDBNodeStoryEnv (_gc_repofilepath cfg)
       pool    <- newPool            $ _gc_database_config cfg
       nodeStory_env <- fromDBNodeStoryEnv pool
-      let setts = devSettings
       pure $ WorkerEnv
         { _w_env_pool      = pool
         , _w_env_logger    = logger
         , _w_env_nodeStory = nodeStory_env
-        , _w_env_settings  = setts
         , _w_env_config    = cfg
         , _w_env_mail      = _gc_mail_config cfg
         , _w_env_nlp       = nlpServerMap $ _gc_nlp_config cfg
@@ -84,9 +80,6 @@ withWorkerEnv settingsFile k = withLoggerHoisted Dev $ \logger -> do
 instance HasConfig WorkerEnv where
   hasConfig = to _w_env_config
 
-instance HasSettings WorkerEnv where
-  settings = to _w_env_settings
-
 instance HasLogger (GargM WorkerEnv IOException) where
   data instance Logger (GargM WorkerEnv IOException)  =
     GargWorkerLogger {

src/Gargantext/Core/Worker/Jobs.hs

@@ -13,43 +13,39 @@ Portability : POSIX
 module Gargantext.Core.Worker.Jobs where
 
 
-import Async.Worker.Broker.Redis (RedisBroker, BrokerInitParams(RedisBrokerInitParams))
-import Async.Worker.Broker.Types (Broker, initBroker)
+import Async.Worker.Broker.PGMQ (PGMQBroker)
 import Async.Worker qualified as Worker
-import Async.Worker.Types qualified as Worker
 import Async.Worker.Types (HasWorkerBroker)
 import Control.Lens (view)
-import Database.Redis qualified as Redis
+import Gargantext.API.Admin.EnvTypes qualified as EnvTypes
 import Gargantext.Core.Config (gc_worker, HasConfig(..))
-import Gargantext.Core.Config.Worker (WorkerSettings(..), WorkerDefinition(..), wdToRedisConnectInfo)
+import Gargantext.Core.Config.Worker (WorkerSettings(..), WorkerDefinition(..))
+import Gargantext.Core.Worker.Broker (initBrokerWithDBCreate)
 import Gargantext.Core.Worker.Jobs.Types (Job(..))
 import Gargantext.Database.Prelude (Cmd')
 import Gargantext.Prelude
 
 
-initializeRedisBroker :: (HasWorkerBroker RedisBroker Job)
-                      => Redis.ConnectInfo
-                      -> IO (Broker RedisBroker (Worker.Job Job))
-initializeRedisBroker connInfo = do
-  let initParams = RedisBrokerInitParams connInfo
-
-  initBroker initParams
-
-
-sendJob :: (HasWorkerBroker RedisBroker Job, HasConfig env)
+sendJob :: (HasWorkerBroker PGMQBroker Job, HasConfig env)
         => Job
         -> Cmd' env err ()
 sendJob job = do
-  ws <- view $ hasConfig . gc_worker
+  gcConfig <- view $ hasConfig
+  let WorkerSettings { _wsDefinitions } = gcConfig ^. gc_worker
   -- TODO Try to guess which worker should get this job
   -- let mWd = findDefinitionByName ws workerName
-  let mWd = head $ _wsDefinitions ws
+  let mWd = head _wsDefinitions
   case mWd of
-    Nothing -> panicTrace $ "worker definition not found"
+    Nothing -> panicTrace "No worker definitions available"
     Just wd -> liftBase $ do
-      case wdToRedisConnectInfo wd of
-        Nothing -> panicTrace $ "worker definition: could not create redis conn info"
-        Just connInfo -> do
-          b <- initializeRedisBroker connInfo
-          let queueName = _wdQueue wd
-          void $ Worker.sendJob' $ Worker.mkDefaultSendJob' b queueName job
+      b <- initBrokerWithDBCreate gcConfig
+      let queueName = _wdQueue wd
+      void $ Worker.sendJob' $ Worker.mkDefaultSendJob' b queueName job
+
+
+-- | This is just a list of what's implemented and what not.
+--   After we migrate to async workers, this should be removed
+--   (see G.C.Worker -> performAction on what's implemented already)
+handledJobs :: [ EnvTypes.GargJob ]
+handledJobs = [ EnvTypes.AddCorpusQueryJob
+              , EnvTypes.ForgotPasswordJob ]

src/Gargantext/Core/Worker/Jobs/Types.hs

+{-|
+Module      : Gargantext.Core.Worker.Jobs.Types
+Description : Worker job definitions
+Copyright   : (c) CNRS, 2024-Present
+License     : AGPL + CECILL v3
+Maintainer  : team@gargantext.org
+Stability   : experimental
+Portability : POSIX
+
+-}
+
+
+module Gargantext.Core.Worker.Jobs.Types where
+
+
+import Data.Aeson ((.:), (.=), object, withObject)
+import Data.Aeson.Types (prependFailure, typeMismatch)
+import Gargantext.API.Admin.Auth.Types (AuthenticatedUser, ForgotPasswordAsyncParams)
+import Gargantext.API.Admin.EnvTypes ( GargJob )
+import Gargantext.API.Node.New.Types ( PostNode(..) )
+import Gargantext.API.Node.Types (NewWithForm, WithQuery)
+import Gargantext.Core.Types.Individu (User)
+import Gargantext.Database.Admin.Types.Node (CorpusId, NodeId)
+import Gargantext.Prelude
+
+
+data Job =
+    Ping
+  | AddCorpusFormAsync { _acf_args :: NewWithForm
+                       , _acf_user :: User
+                       , _acf_cid  :: CorpusId }
+  | AddCorpusWithQuery { _acq_args :: WithQuery
+                       , _acq_user :: User
+                       , _acq_cid   :: CorpusId }
+  | ForgotPasswordAsync { _fpa_args :: ForgotPasswordAsyncParams }
+  | NewNodeAsync { _nna_node_id :: NodeId
+                 , _nna_authenticatedUser :: AuthenticatedUser
+                 , _nna_postNode :: PostNode }
+  | GargJob { _gj_garg_job :: GargJob }
+  deriving (Show, Eq)
+instance FromJSON Job where
+  parseJSON = withObject "Job" $ \o -> do
+    type_ <- o .: "type"
+    case type_ of
+      "Ping" -> return Ping
+      "AddCorpusFormAsync" -> do
+        _acf_args <- o .: "args"
+        _acf_user <- o .: "user"
+        _acf_cid <- o .: "cid"
+        return $ AddCorpusFormAsync { .. }
+      "AddCorpusWithQuery" -> do
+        _acq_args <- o .: "args"
+        _acq_user <- o .: "user"
+        _acq_cid <- o .: "cid"
+        return $ AddCorpusWithQuery { .. }
+      "ForgotPasswordAsync" -> do
+        _fpa_args <- o .: "args"
+        return $ ForgotPasswordAsync { _fpa_args }
+      "NewNodeAsync" -> do
+        _nna_node_id <- o .: "node_id"
+        _nna_authenticatedUser <- o .: "authenticated_user"
+        _nna_postNode <- o .: "post_node"
+        return $ NewNodeAsync { .. }
+      "GargJob" -> do
+        _gj_garg_job <- o .: "garg_job"
+        return $ GargJob { _gj_garg_job }
+      s -> prependFailure "parsing job type failed, " (typeMismatch "type" s)
+instance ToJSON Job where
+  toJSON Ping = object [ ("type" .= ("Ping" :: Text)) ]
+  toJSON (AddCorpusFormAsync { .. }) =
+    object [ ("type" .= ("AddCorpusFormJob" :: Text))
+           , ("args" .= _acf_args)
+           , ("user" .= _acf_user)
+           , ("cid" .= _acf_cid) ]
+  toJSON (AddCorpusWithQuery { .. }) =
+    object [ ("type" .= ("AddCorpusWithQuery" :: Text))
+           , ("args" .= _acq_args)
+           , ("user" .= _acq_user)
+           , ("cid" .= _acq_cid) ]
+  toJSON (ForgotPasswordAsync { .. }) =
+    object [ ("type" .= ("ForgotPasswordAsync" :: Text))
+           , ("args" .= _fpa_args) ]
+  toJSON (NewNodeAsync { .. }) =
+    object [ ("type" .= ("NewNodeAsync" :: Text))
+           , ("node_id" .= _nna_node_id)
+           , ("authenticated_user" .= _nna_authenticatedUser)
+           , ("post_node" .= _nna_postNode) ]
+  toJSON (GargJob { .. }) =
+    object [ ("type" .= ("GargJob" :: Text))
+           , ("garg_job" .= _gj_garg_job) ]

src/Gargantext/Database/Query/Table/Node/Update.hs

@@ -14,7 +14,7 @@ module Gargantext.Database.Query.Table.Node.Update (Update(..), update)
 
 import Data.Text qualified as DT
 import Database.PostgreSQL.Simple ( Only(Only) )
-import Gargantext.Core.Notifications.CentralExchange.Types qualified as CE
+import Gargantext.Core.AsyncUpdates.CentralExchange.Types qualified as CE
 import Gargantext.Core.Types (Name)
 import Gargantext.Database.Admin.Types.Node (NodeId, ParentId)
 import Gargantext.Database.Query.Table.Node (getParentId)

src/Gargantext/Utils/Jobs.hs

@@ -22,22 +22,20 @@ module Gargantext.Utils.Jobs (
   , markFailedNoErr
   ) where
 
-import Control.Monad.Except ( runExceptT )
-import Control.Monad.Reader ( MonadReader(ask), ReaderT(runReaderT) )
-import Data.Aeson (ToJSON)
-import Prelude
-import System.Directory (doesFileExist)
-import Text.Read (readMaybe)
-import qualified Data.Text as T
-
-import Gargantext.API.Admin.EnvTypes ( mkJobHandle, Env, GargJob(..) )
+import Data.Text qualified as T
+import Gargantext.API.Admin.EnvTypes ( mkJobHandle, parseGargJob, Env, GargJob(..) )
 import Gargantext.API.Errors.Types ( BackendInternalError(InternalJobError) )
 import Gargantext.API.Prelude ( GargM )
-import qualified Gargantext.Utils.Jobs.Internal as Internal
-import Gargantext.Utils.Jobs.Monad ( JobError, MonadJobStatus(..), markFailureNoErr, markFailedNoErr )
+import Gargantext.Core.Worker.Jobs qualified as Jobs
+import Gargantext.Core.Worker.Jobs.Types qualified as Jobs
+import Gargantext.Prelude
 import Gargantext.System.Logging
+import Gargantext.Utils.Jobs.Internal qualified as Internal
+import Gargantext.Utils.Jobs.Monad ( JobError, MonadJobStatus(..), markFailureNoErr, markFailedNoErr )
+-- import Prelude
+import Servant.Job.Async qualified as SJ
+import System.Directory (doesFileExist)
 
-import qualified Servant.Job.Async as SJ
 
 jobErrorToGargError
   :: JobError -> BackendInternalError
@@ -61,38 +59,21 @@ serveJobsAPI
 serveJobsAPI jobType f = Internal.serveJobsAPI mkJobHandle ask jobType jobErrorToGargError $ \env jHandle i -> do
   runExceptT $ flip runReaderT env $ do
     $(logLocM) INFO (T.pack $ "Running job of type: " ++ show jobType)
+    unless (jobType `elem` Jobs.handledJobs) $
+      Jobs.sendJob $ Jobs.GargJob { Jobs._gj_garg_job = jobType }
     f jHandle i
     getLatestJobStatus jHandle
 
-parseGargJob :: String -> Maybe GargJob
-parseGargJob s = case s of
-  "tablengrams" -> Just TableNgramsJob
-  "forgotpassword" -> Just ForgotPasswordJob
-  "updatengramslistjson" -> Just UpdateNgramsListJobJSON
-  "updatengramslisttsv" -> Just UpdateNgramsListJobTSV
-  "addcontact" -> Just AddContactJob
-  "addfile" -> Just AddFileJob
-  "documentfromwritenode" -> Just DocumentFromWriteNodeJob
-  "updatenode" -> Just UpdateNodeJob
-  "updateframecalc" -> Just UploadFrameCalcJob
-  "updatedocument" -> Just UploadDocumentJob
-  "newnode" -> Just NewNodeJob
-  "addcorpusquery" -> Just AddCorpusQueryJob
-  "addcorpusform" -> Just AddCorpusFormJob
-  "addcorpusfile" -> Just AddCorpusFileJob
-  "addannuaireform" -> Just AddAnnuaireFormJob
-  "recomputegraph" -> Just RecomputeGraphJob
-  _ -> Nothing
-
-parsePrios :: [String] -> IO [(GargJob, Int)]
+parsePrios :: [Text] -> IO [(GargJob, Int)]
 parsePrios []       = pure []
-parsePrios (x : xs) = (:) <$> go x <*> parsePrios xs
-  where go s = case break (=='=') s of
-          ([], _) -> error "parsePrios: empty jobname?"
+parsePrios (x : xs) = (:) <$> go (T.unpack x) <*> parsePrios xs
+  where
+    go s = case break (=='=') s of
+          ([], _) -> errorTrace "parsePrios: empty jobname?"
           (prop, valS)
-            | Just val <- readMaybe (tail valS)
-            , Just j   <- parseGargJob prop -> pure (j, val)
-            | otherwise                  -> error $
+            | Just val <- readMaybe (T.tail $ T.pack valS)
+            , Just j   <- parseGargJob (T.pack prop) -> pure (j, val)
+            | otherwise                  -> errorTrace $
               "parsePrios: invalid input. " ++ show (prop, valS)
 
 readPrios :: Logger IO -> FilePath -> IO [(GargJob, Int)]

src/Gargantext/Utils/Jobs/Internal.hs

+{-|
+Module      : Gargantext.Utils.Jobs.Internal
+Description : Servant Jobs
+Copyright   : (c) CNRS, 2024-Present
+License     : AGPL + CECILL v3
+Maintainer  : team@gargantext.org
+Stability   : experimental
+Portability : POSIX
+
+-}
+
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE ViewPatterns #-}
+
 module Gargantext.Utils.Jobs.Internal (
     serveJobsAPI
   -- * Internals for testing

stack.yaml

@@ -17,6 +17,7 @@
 - "binary-orphans-1.0.5"
 - "blaze-html-0.9.2.0"
 - "boring-0.2.2"
+- "bytestring-lexing-0.5.0.14"
 - "bzlib-conduit-0.3.0.3"
 - "cabal-doctest-1.0.10"
 - "cassava-0.5.3.2"
@@ -62,6 +63,7 @@
 - "language-c-0.9.3"
 - "libyaml-0.1.4"
 - "libyaml-clib-0.2.5"
+- "linear-1.23"
 - "logict-0.8.1.0"
 - "lzma-0.0.1.1"
 - "math-functions-0.3.4.4"
@@ -127,6 +129,7 @@
 - "type-equality-1.0.1"
 - "typed-process-0.2.12.0"
 - "unicode-collation-0.1.3.6"
+- "units-2.4.1.5"
 - "unix-compat-0.7.2"
 - "unix-time-0.4.15"
 - "unordered-containers-0.2.20"
@@ -257,6 +260,10 @@
   git: "https://gitlab.iscpif.fr/gargantext/gargantext-graph.git"
   subdirs:
   - .
+- commit: 58ab07e0110281f94ecc8840b8cd0c0a9081b672
+  git: "https://gitlab.iscpif.fr/gargantext/haskell-bee"
+  subdirs:
+  - .
 - commit: bb15d828d5ef36eeaa84cccb00598b585048c88e
   git: "https://gitlab.iscpif.fr/gargantext/haskell-gargantext-prelude"
   subdirs:
@@ -269,6 +276,10 @@
   git: "https://gitlab.iscpif.fr/gargantext/haskell-infomap.git"
   subdirs:
   - .
+- commit: 0591a643d8ba1776af4fac56c1e4ff5fc3e98bb3
+  git: "https://gitlab.iscpif.fr/gargantext/haskell-pgmq"
+  subdirs:
+  - .
 - commit: 02f5ed9ee2d6cce45161addf945b88bc6adf9059
   git: "https://gitlab.iscpif.fr/gargantext/haskell-throttle"
   subdirs:
@@ -301,6 +312,8 @@ flags:
     "warp-tests": false
   JuicyPixels:
     mmap: false
+  MemoTrie:
+    examples: false
   MissingH:
     "network--ge-3_0_0": true
   QuickCheck:
@@ -436,6 +449,8 @@ flags:
     portable: false
     sse42: false
     "unsafe-tricks": true
+  hedis:
+    dev: false
   hmatrix:
     "disable-default-paths": false
     "no-random_r": false
@@ -487,6 +502,9 @@ flags:
   libyaml:
     "no-unicode": false
     "system-libyaml": false
+  linear:
+    herbie: false
+    "template-haskell": true
   "llvm-hs":
     debug: false
     "llvm-with-rtti": false

test-data/test_config.toml

@@ -72,3 +72,15 @@ dispatcher = { bind = "tcp://*:15561", connect = "tcp://localhost:15561" }
 EN = "corenlp://localhost:9000"
 FR = "spacy://localhost:8001"
 All = "corenlp://localhost:9000"
+
+[worker]
+[worker.database]
+host = "127.0.0.1"
+port = 5432
+name = "pgmq_test"
+user = "gargantua"
+pass = "gargantua_test"
+
+[[worker.definitions]]
+name = "default"
+queue = "default"

test/Test/Core/Worker.hs

+{-|
+Module      : Test.Core.Worker
+Description :
+Copyright   : (c) CNRS, 2024-Present
+License     : AGPL + CECILL v3
+Maintainer  : team@gargantext.org
+Stability   : experimental
+Portability : POSIX
+
+-}
+
+module Test.Core.Worker where
+
+import Data.Aeson qualified as Aeson
+import Gargantext.Core.Methods.Similarities.Conditional
+import Gargantext.Core.Worker.Jobs.Types (Job(..))
+import Gargantext.Prelude
+import Test.Instances ()
+import Test.Tasty
+import Test.Tasty.HUnit
+import Test.Tasty.QuickCheck hiding (Positive, Negative)
+
+
+
+tests :: TestTree
+tests = testGroup "worker unit tests" [
+  testProperty "Worker Job to/from JSON serialization is correct" $
+    \job -> Aeson.decode (Aeson.encode (job :: Job)) == Just job
+  ]

test/Test/Instances.hs

@@ -22,21 +22,105 @@ import Data.Patch.Class (Replace(Keep), replace)
 import Data.Text qualified as T
 import Data.Validity (Validation(..), ValidationChain (..), prettyValidation)
 import EPO.API.Client.Types qualified as EPO
+import Gargantext.API.Admin.Auth.Types (AuthenticatedUser(..), ForgotPasswordAsyncParams(..))
+import Gargantext.API.Admin.EnvTypes as EnvTypes
 import Gargantext.API.Errors.Types qualified as Errors
 import Gargantext.API.Ngrams.Types qualified as Ngrams
 import Gargantext.API.Node.Corpus.New (ApiInfo(..))
-import Gargantext.API.Node.Types (RenameNode(..), WithQuery(..))
-import Gargantext.Core.Notifications.CentralExchange.Types qualified as CET
-import Gargantext.Core.Notifications.Dispatcher.Types qualified as DET
+import Gargantext.API.Node.Corpus.New qualified as New
+import Gargantext.API.Node.Types (NewWithForm(..), RenameNode(..), WithQuery(..))
+import Gargantext.Core.AsyncUpdates.CentralExchange.Types qualified as CET
+import Gargantext.Core.AsyncUpdates.Dispatcher.Types qualified as DET
 import Gargantext.Core.NodeStory.Types qualified as NS
 import Gargantext.Core.Text.Ngrams (NgramsType(..))
+import Gargantext.Core.Types.Individu qualified as Individu
 import Gargantext.Core.Types.Main (ListType(CandidateTerm, StopTerm, MapTerm))
+import Gargantext.Core.Worker.Jobs.Types (Job(..))
 import Gargantext.Database.Admin.Types.Node (UserId(UnsafeMkUserId))
 import Gargantext.Database.Admin.Types.Hyperdata qualified as Hyperdata
 import Gargantext.Prelude hiding (replace, Location)
 import Servant.Job.Core qualified as SJ
 import Servant.Job.Types qualified as SJ
 import Test.QuickCheck
+import Test.Tasty.QuickCheck hiding (Positive, Negative)
+import Text.Parsec.Error (ParseError, Message(..), newErrorMessage)
+import Text.Parsec.Pos
+
+
+instance Arbitrary AuthenticatedUser where
+  arbitrary = AuthenticatedUser <$> arbitrary  -- _auth_node_id
+                                <*> arbitrary  -- _auth_user_id
+
+
+instance Arbitrary EnvTypes.GargJob where
+  arbitrary = do
+    oneof [ pure AddAnnuaireFormJob
+          , pure AddContactJob
+          , pure AddCorpusFileJob
+          , pure AddCorpusFormJob
+          , pure AddCorpusQueryJob
+          , pure AddFileJob
+          , pure DocumentFromWriteNodeJob
+          , pure ForgotPasswordJob
+          , pure NewNodeJob
+          , pure RecomputeGraphJob
+          , pure TableNgramsJob
+          , pure UpdateNgramsListJobJSON
+          , pure UpdateNgramsListJobTSV
+          , pure UpdateNodeJob
+          , pure UploadDocumentJob
+          , pure UploadFrameCalcJob
+          ]
+
+
+instance Arbitrary Job where
+  arbitrary = oneof [ pure Ping
+                    , addCorpusFormAsyncGen
+                    , forgotPasswordAsyncGen
+                    , newNodeAsyncGen
+                    , gargJobGen ]
+    where
+      forgotPasswordAsyncGen = do
+        email <- arbitrary
+        return $ ForgotPasswordAsync (ForgotPasswordAsyncParams { email })
+      addCorpusFormAsyncGen = do
+        _acf_args <- arbitrary
+        _acf_user <- arbitrary
+        _acf_cid <- arbitrary
+        return $ AddCorpusFormAsync { .. }
+      newNodeAsyncGen = do
+        _nna_node_id <- arbitrary
+        _nna_authenticatedUser <- arbitrary
+        _nna_postNode <- arbitrary
+        return $ NewNodeAsync { .. }
+      gargJobGen = do
+        _gj_garg_job <- arbitrary
+        return $ GargJob { _gj_garg_job }
+
+
+
+
+instance Arbitrary Message where
+  arbitrary = do
+    msgContent <- arbitrary
+    oneof $ return <$> [SysUnExpect msgContent
+                       , UnExpect msgContent
+                       , Expect msgContent
+                       , Message msgContent
+                       ]
+
+instance Arbitrary SourcePos where
+  arbitrary = do
+    sn <- arbitrary
+    l <- arbitrary
+    c <- arbitrary
+    return $ newPos sn l c
+
+instance Arbitrary ParseError where
+  arbitrary = do
+    sp <- arbitrary
+    msg <- arbitrary
+    return $ newErrorMessage msg sp
 
 
 
@@ -53,6 +137,16 @@ alphanum :: [Char]
 alphanum = smallLetter <> largeLetter <> digit
 
 
+instance Arbitrary Individu.User where
+  arbitrary = do
+    userId <- arbitrary
+    userName <- arbitrary
+    nodeId <- arbitrary
+    oneof [ pure $ Individu.UserDBId userId
+          , pure $ Individu.UserName userName
+          , pure $ Individu.RootId nodeId ]
+
+
 instance Arbitrary EPO.AuthKey where
   arbitrary = do
     user <- arbitrary
@@ -104,6 +198,14 @@ instance Arbitrary Hyperdata.HyperdataPublic where
 instance Arbitrary a => Arbitrary (SJ.JobOutput a) where
   arbitrary = SJ.JobOutput <$> arbitrary
 
+instance Arbitrary NewWithForm where
+  arbitrary = NewWithForm <$> arbitrary  -- _wf_filetype
+                          <*> arbitrary  -- _wf_fileformat
+                          <*> arbitrary  -- _wf_data
+                          <*> arbitrary  -- _wf_lang
+                          <*> arbitrary  -- _wf_name
+                          <*> arbitrary  -- _wf_selection
+
 instance Arbitrary RenameNode where
   arbitrary = elements [RenameNode "test"]
   

test/Test/Parsers/Types.hs

@@ -19,6 +19,7 @@ module Test.Parsers.Types where
 
 import Gargantext.Prelude
 
+import Test.Instances ()
 import Test.QuickCheck
 import Test.QuickCheck.Instances ()
 
@@ -43,25 +44,3 @@ looseZonedTimePrecision (ZonedTime lt tz) = ZonedTime (looseLocalTimePrecision l
 loosePrecisionEitherPEZT :: Either ParseError ZonedTime -> Either ParseError ZonedTime
 loosePrecisionEitherPEZT (Right zt) = Right $ looseZonedTimePrecision zt
 loosePrecisionEitherPEZT pe = pe
-
-instance Arbitrary Message where
-  arbitrary = do
-    msgContent <- arbitrary
-    oneof $ return <$> [SysUnExpect msgContent
-                       , UnExpect msgContent
-                       , Expect msgContent
-                       , Message msgContent
-                       ]
-
-instance Arbitrary SourcePos where
-  arbitrary = do
-    sn <- arbitrary
-    l <- arbitrary
-    c <- arbitrary
-    return $ newPos sn l c
-
-instance Arbitrary ParseError where
-  arbitrary = do
-    sp <- arbitrary
-    msg <- arbitrary
-    return $ newErrorMessage msg sp

test/Test/Utils/Jobs.hs

@@ -12,12 +12,13 @@ Portability : POSIX
 {-# LANGUAGE TypeApplications    #-}
 {-# LANGUAGE TypeFamilies        #-}
 
-module Test.Utils.Jobs (test) where
+module Test.Utils.Jobs ( test, qcTests ) where
 
 import Control.Concurrent
 import Control.Concurrent.Async
 import Control.Concurrent.Async qualified as Async
 import Control.Concurrent.STM
+import Data.Aeson qualified as Aeson
 import Data.Sequence ((|>), fromList)
 import Data.Time
 import Debug.RecoverRTTI (anythingToString)
@@ -43,6 +44,9 @@ import System.IO.Unsafe
 import System.Timeout (timeout)
 import Test.Hspec
 import Test.Hspec.Expectations.Contrib (annotate)
+import Test.Instances ()  -- arbitrary instances
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.QuickCheck hiding (Positive, Negative)
 import Test.Utils (waitUntil)
 
 
@@ -485,3 +489,10 @@ test = do
       testFetchJobStatusNoContention
     it "marking stuff behaves as expected" $
       testMarkProgress
+
+
+qcTests :: TestTree
+qcTests = testGroup "jobs qc tests" [
+  testProperty "GargJob to/from JSON serialization is correct" $
+    \job -> Aeson.decode (Aeson.encode (job :: EnvTypes.GargJob)) == Just job
+  ]

test/drivers/tasty/Main.hs

@@ -15,6 +15,7 @@ import Gargantext.Prelude
 import qualified Test.Core.Text.Corpus.Query     as CorpusQuery
 import qualified Test.Core.Text.Corpus.TSV       as TSVParser
 import qualified Test.Core.Utils                 as Utils
+import qualified Test.Core.Worker                as Worker
 import qualified Test.Graph.Clustering           as Graph
 import qualified Test.Ngrams.NLP                 as NLP
 import qualified Test.Ngrams.Query               as NgramsQuery
@@ -57,6 +58,8 @@ main = do
     , similaritySpec
     , Phylo.tests
     , testGroup "Stemming" [ Lancaster.tests ]
+    , Worker.tests
+    , Jobs.qcTests
     , asyncUpdatesSpec
     , Notifications.qcTests
     ]