[workers] this is still so much WIP

src/Gargantext/AsyncTasks.purs

 module Gargantext.AsyncTasks (
-    TaskList
+    Task
+  , TaskList
   , Storage(..)
   , readAsyncTasks
   , insert
@@ -33,7 +34,9 @@ import Toestand as T
 import Web.Storage.Storage as WSS
 
 
-type TaskList = Array GT.AsyncTaskWithType
+-- type TaskList = Array GT.AsyncTaskWithType
+type Task = GT.WorkerTask
+type TaskList = Array Task
 newtype Storage = Storage (Map.Map GT.NodeID TaskList)
 derive newtype instance Semigroup Storage
 derive newtype instance Monoid Storage
@@ -83,9 +86,13 @@ setTasks id tasks (Storage s) = Storage $ Map.insert id tasks s
 focus :: GT.NodeID -> T.Box Storage -> R.Hooks (T.Box TaskList)
 focus id tasks = T.useFocused (getTasks id) (setTasks id) tasks
 
-removeTaskFromList :: TaskList -> GT.AsyncTaskWithType -> TaskList
-removeTaskFromList ts (GT.AsyncTaskWithType { task: GT.AsyncTask { id: id' } }) =
-  A.filter (\(GT.AsyncTaskWithType { task: GT.AsyncTask { id: id'' } }) -> id' /= id'') ts
+-- removeTaskFromList :: TaskList -> GT.AsyncTaskWithType -> TaskList
+-- removeTaskFromList ts (GT.AsyncTaskWithType { task: GT.AsyncTask { id: id' } }) =
+--   A.filter (\(GT.AsyncTaskWithType { task: GT.AsyncTask { id: id'' } }) -> id' /= id'') ts
+
+removeTaskFromList :: TaskList -> Task -> TaskList
+removeTaskFromList ts (GT.WorkerTask { message_id }) =
+  A.filter (\(GT.WorkerTask { message_id: message_id' }) -> message_id /= message_id') ts
 
 type ReductorProps = (
     reloadForest :: T2.ReloadS
@@ -93,15 +100,15 @@ type ReductorProps = (
   , storage      :: Storage
   )
 
-insert :: GT.NodeID -> GT.AsyncTaskWithType -> T.Box Storage -> Effect Unit
+insert :: GT.NodeID -> Task -> T.Box Storage -> Effect Unit
 insert id task storageBox = modifyTaskBox newStorage storageBox
   where
-    newStorage (Storage s) = Storage $ Map.alter (maybe (Just [task]) (\ts -> Just $ A.cons task ts)) id s
+    newStorage (Storage s) = Storage $ Map.alter (maybe (Just [task]) (\ts -> Just $ A.nub $ A.cons task ts)) id s
 
-finish :: GT.NodeID -> GT.AsyncTaskWithType -> T.Box Storage -> Effect Unit
+finish :: GT.NodeID -> Task -> T.Box Storage -> Effect Unit
 finish id task storage = remove id task storage
 
-remove :: GT.NodeID -> GT.AsyncTaskWithType -> T.Box Storage -> Effect Unit
+remove :: GT.NodeID -> Task -> T.Box Storage -> Effect Unit
 remove id task storageBox = modifyTaskBox newStorage storageBox
   where
     newStorage (Storage s) = Storage $ Map.alter (maybe Nothing $ (\ts -> Just $ removeTaskFromList ts task)) id s

src/Gargantext/Components/FolderView.purs

@@ -372,7 +372,8 @@ performAction = performAction' where
   updateNode params { boxes: { errors, tasks }, nodeId: id, session } = do
     eTask <- updateRequest params session id
     handleRESTError (R2.herePrefix here "[updateNode]") errors eTask $ \task -> liftEffect $ do
-      GAT.insert id task tasks
+      -- GAT.insert id task tasks
+      here.log "[performAction] TODO: IMPLEMENT ME!"
 
   sharePublic params p@{ boxes: { errors }, session } = traverse_ f params where
     f (SubTreeOut { in: inId, out }) = do
@@ -386,12 +387,14 @@ performAction = performAction' where
   uploadFile' nodeType fileType fileFormat lang mName contents { boxes: { errors, tasks }, nodeId: id, session } selection = do
     eTask <- uploadFile { contents, fileType, fileFormat, lang, id, nodeType, mName, selection, session }
     handleRESTError (R2.herePrefix here "[uploadFile']") errors eTask $ \task -> liftEffect $ do
-      GAT.insert id task tasks
+      -- GAT.insert id task tasks
+      here.log "[performAction] TODO: IMPLEMENT ME!"
 
   uploadArbitraryFile' fileFormat mName blob { boxes: { errors, tasks }, nodeId: id, session } = do
     eTask <- uploadArbitraryFile session id { blob, fileFormat, mName }
     handleRESTError (R2.herePrefix here "[uploadArbitraryFile']") errors eTask $ \task -> liftEffect $ do
-      GAT.insert id task tasks
+      -- GAT.insert id task tasks
+      here.log "[performAction] TODO: IMPLEMENT ME!"
 
   moveNode params p@{ boxes: { errors }, session } = traverse_ f params where
     f (SubTreeOut { in: in', out }) = do

src/Gargantext/Components/Forest/Tree.purs

@@ -30,7 +30,7 @@ import Gargantext.Components.Forest.Tree.Node.Action.WriteNodesDocuments (docume
 import Gargantext.Components.Forest.Tree.Node.Tools.FTree (FTree, LNode(..), NTree(..), fTreeID)
 import Gargantext.Components.Forest.Tree.Node.Tools.SubTree.Types (SubTreeOut(..))
 import Gargantext.Components.Notifications as Notifications
-import Gargantext.Components.Notifications.Types as NotificationsT
+import Gargantext.Components.Notifications.Types as NT
 import Gargantext.Config.REST (AffRESTError, logRESTError)
 import Gargantext.Config.Utils (handleRESTError)
 import Gargantext.Ends (Frontends)
@@ -181,12 +181,17 @@ treeCpt = here.component "tree" cpt where
       -- rendered via 'childLoader'. However, we still need a hook
       -- here, so that if e.g. the tree is pinned, it becomes its own
       -- root and we want to see notifications of it as well.
-      let cb _ = do
-            here.log2 "[tree] callback!" root
-            -- The modal window has some problems closing when we refresh too early. This is a HACK
-            void $ setTimeout 400 $ T2.reload reload
-      let action = NotificationsT.InsertCallback (NotificationsT.UpdateTree root) ("tree-" <> show root) cb
+      let cb n = do
+            case n of
+              NT.NUpdateTree _ -> do
+                here.log2 "[tree] update tree" root
+                -- The modal window has some problems closing when we refresh too early. This is a HACK
+                void $ setTimeout 400 $ T2.reload reload
+              NT.NUpdateWorkerProgress ji jl -> do
+                here.log3 "[tree] update worker progress" ji jl
+              _ -> pure unit
       ws <- T.read boxes.wsNotification
+      let action = NT.InsertCallback (NT.UpdateTree root) ("tree-" <> show root) cb
       Notifications.performAction ws action
 
     R.useEffect' do
@@ -337,12 +342,15 @@ childLoaderCpt = R2.hereComponent here "childLoader" hCpt where
 
     boxes <- Store.use
     R.useEffectOnce' $ do
-      let cb _ = do
-            here.log2 "[childLoader] callback!" p.id
-            -- The modal window has some problems closing when we refresh too early. This is a HACK
-            void $ setTimeout 400 $ T2.reload reload
-      let action = NotificationsT.InsertCallback (NotificationsT.UpdateTree p.id) ("tree-" <> show p.id) cb
+      let cb n = do
+            case n of
+              NT.NUpdateTree _nId -> do
+                here.log2 "[childLoader] update tree" p.id
+                -- The modal window has some problems closing when we refresh too early. This is a HACK
+                void $ setTimeout 400 $ T2.reload reload
+              _ -> pure unit
       ws <- T.read boxes.wsNotification
+      let action = NT.InsertCallback (NT.UpdateTree p.id) ("tree-" <> show p.id) cb
       Notifications.performAction ws action
 
     -- Render
@@ -384,7 +392,8 @@ doSearch task { boxes: { tasks }, tree: NTree (LNode {id}) _ } = liftEffect $ do
 updateNode params p@{ boxes: { errors, tasks }, session, tree: (NTree (LNode {id}) _) } = do
   eTask <- updateRequest params session id
   handleRESTError (R2.herePrefix here "[updateNode]") errors eTask $ \task -> liftEffect $ do
-    GAT.insert id task tasks
+    -- GAT.insert id task tasks
+    here.log "[updateNode] TODO: IMPLEMENT ME!"
     closeBox p
 
 renameNode name p@{ boxes: { errors }, session, tree: (NTree (LNode {id}) _) } = do
@@ -412,19 +421,22 @@ addNode' name nodeType p@{ boxes: { errors, forestOpen }, session, tree: (NTree
 uploadFile' nodeType fileType fileFormat lang mName contents p@{ boxes: { errors, tasks }, session, tree: (NTree (LNode { id }) _) } selection = do
   eTask <- uploadFile { contents, fileFormat, fileType, id, lang, mName, nodeType, selection, session }
   handleRESTError (R2.herePrefix here "[uploadFile']") errors eTask $ \task -> liftEffect $ do
-    GAT.insert id task tasks
+    -- GAT.insert id task tasks
+    here.log "[uplaodFile'] TODO: IMPLEMENT ME!"
     here.log2 "[uploadFile'] UploadFile, uploaded, task:" task
     closeBox p
 
 uploadArbitraryFile' fileFormat mName blob { boxes: { errors, tasks }, session, tree: (NTree (LNode { id }) _) } = do
   eTask <- uploadArbitraryFile session id { blob, fileFormat, mName }
   handleRESTError (R2.herePrefix here "[uploadArbitraryFile']") errors eTask $ \task -> liftEffect $ do
-    GAT.insert id task tasks
+    -- GAT.insert id task tasks
+    here.log "[uplaodArbitraryFile'] TODO: IMPLEMENT ME!"
 
 uploadFrameCalc' lang { boxes: { errors, tasks }, session, tree: (NTree (LNode { id }) _) } selection = do
   eTask <- uploadFrameCalc session id lang selection
   handleRESTError (R2.herePrefix here "[uploadFrameCalc']") errors eTask $ \task -> liftEffect $ do
-    GAT.insert id task tasks
+    -- GAT.insert id task tasks
+    here.log "[uplaodFrameCalc'] TODO: IMPLEMENT ME!"
 
 moveNode params p@{ boxes: { errors, forestOpen }, session } = traverse_ f params where
   f (SubTreeOut { in: in', out }) = do
@@ -448,7 +460,8 @@ linkNode nodeType params p@{ boxes: { errors }, session } = traverse_ f params w
 documentsFromWriteNodes params p@{ boxes: { errors, tasks }, session, tree: NTree (LNode { id }) _ } = do
   eTask <- documentsFromWriteNodesReq session params
   handleRESTError (R2.herePrefix here "[documentsFromWriteNodes]") errors eTask $ \task -> liftEffect $ do
-    GAT.insert id task tasks
+    -- GAT.insert id task tasks
+    here.log "[documentsFromWriteNodes] TODO: IMPLEMENT ME!"
     pure unit
   refreshTree p
 

src/Gargantext/Components/Forest/Tree/Node.purs

@@ -14,6 +14,7 @@ import Data.String.Regex as Regex
 import Effect (Effect)
 import Effect.Aff (Aff, launchAff)
 import Effect.Class (liftEffect)
+import Effect.Timer (setTimeout)
 import Gargantext.AsyncTasks as GAT
 import Gargantext.Components.App.Store as AppStore
 import Gargantext.Components.Bootstrap as B
@@ -28,6 +29,8 @@ import Gargantext.Components.Forest.Tree.Node.Tools.Sync (nodeActionsGraph, node
 import Gargantext.Components.GraphExplorer.API as GraphAPI
 import Gargantext.Components.Lang (Lang(EN))
 import Gargantext.Components.Nodes.Corpus.Types (CorpusData)
+import Gargantext.Components.Notifications as Notifications
+import Gargantext.Components.Notifications.Types as NT
 import Gargantext.Context.Progress (asyncContext, asyncProgress)
 import Gargantext.Ends (Frontends, url)
 import Gargantext.Hooks.Loader (useLoader)
@@ -174,27 +177,29 @@ nodeSpanCpt = here.component "nodeSpan" cpt
 
         onTaskFinish ::
              GT.NodeID
-          -> GT.AsyncTaskWithType
+          -> GAT.Task
           -> Unit
           -> Effect Unit
         onTaskFinish id' t _ = do
           GAT.finish id' t boxes.tasks
-          if GAT.asyncTaskTTriggersAppReload t then do
-            here.log2 "reloading root for task" t
-            T2.reload boxes.reloadRoot
-          else do
-            if GAT.asyncTaskTTriggersTreeReload t then do
-              here.log2 "reloading tree for task" t
-              T2.reload reload
-            else do
-              here.log2 "task doesn't trigger a tree reload" t
-              pure unit
-            if GAT.asyncTaskTTriggersMainPageReload t then do
-              here.log2 "reloading main page for task" t
-              T2.reload boxes.reloadMainPage
-            else do
-              here.log2 "task doesn't trigger a main page reload" t
-              pure unit
+          -- TODO App reload!
+          here.log "[onTaskFinish] TODO APP/TREE/MAIN PAGE RELOAD"
+          -- if GAT.asyncTaskTTriggersAppReload t then do
+          --   here.log2 "reloading root for task" t
+          --   T2.reload boxes.reloadRoot
+          -- else do
+          --   if GAT.asyncTaskTTriggersTreeReload t then do
+          --     here.log2 "reloading tree for task" t
+          --     T2.reload reload
+          --   else do
+          --     here.log2 "task doesn't trigger a tree reload" t
+          --     pure unit
+          --   if GAT.asyncTaskTTriggersMainPageReload t then do
+          --     here.log2 "reloading main page for task" t
+          --     T2.reload boxes.reloadMainPage
+          --   else do
+          --     here.log2 "task doesn't trigger a main page reload" t
+          --     pure unit
           -- snd tasks $ GAT.Finish id' t
           -- mT <- T.read tasks
           -- case mT of
@@ -230,6 +235,23 @@ nodeSpanCpt = here.component "nodeSpan" cpt
             else
               H.div {} []
 
+    -- Notifications
+      R.useEffectOnce' $ do
+        let cb n = do
+              case n of
+                NT.NUpdateTree _ -> do
+                  here.log2 "[nodeSpan] update tree" props.id
+                  -- The modal window has some problems closing when we refresh too early. This is a HACK
+                  void $ setTimeout 400 $ T2.reload reload
+                NT.NUpdateWorkerProgress ji jl -> do
+                  -- TODO Fire this only once!
+                  here.log3 "[nodeSpan] update job progress" ji jl
+                  GAT.insert props.id ji boxes.tasks
+                _ -> pure unit
+        ws <- T.read boxes.wsNotification
+        let action = NT.InsertCallback (NT.UpdateTree props.id) ("node-span-" <> show props.id) cb
+        Notifications.performAction ws action
+
     -- Render
 
       pure $

src/Gargantext/Components/Forest/Tree/Node/Action/Search.purs

@@ -3,6 +3,7 @@ module Gargantext.Components.Forest.Tree.Node.Action.Search where
 import Data.Maybe (Maybe(..))
 import Effect (Effect)
 import Effect.Aff (Aff, launchAff)
+import Gargantext.AsyncTasks as GAT
 import Gargantext.Components.App.Store as Store
 import Gargantext.Components.Forest.Tree.Node.Action.Search.SearchBar (searchBar)
 import Gargantext.Components.Forest.Tree.Node.Action.Search.SearchField (defaultSearch)
@@ -68,7 +69,7 @@ actionSearchWithLangsCpt = here.component "actionSearchWithLangs" cpt
         ]
         where
           searchOn :: (Action -> Aff Unit)
-                   -> GT.AsyncTaskWithType
+                   -> GAT.Task
                    -> Effect Unit
           searchOn dispatch' task = do
             _ <- launchAff $ dispatch' (DoSearch task)

src/Gargantext/Components/Forest/Tree/Node/Action/Search/SearchBar.purs

@@ -4,6 +4,7 @@ module Gargantext.Components.Forest.Tree.Node.Action.Search.SearchBar
   ) where
 
 import Effect (Effect)
+import Gargantext.AsyncTasks as GAT
 import Gargantext.Components.Forest.Tree.Node.Action.Search.SearchField (searchField)
 import Gargantext.Components.Forest.Tree.Node.Action.Search.Types (Search, allDatabases)
 import Gargantext.Components.Lang (Lang)
@@ -21,7 +22,7 @@ here = R2.here "Gargantext.Components.Forest.Tree.Node.Action.Search.SearchBar"
 
 type Props = ( errors    :: T.Box (Array FrontendError)
              , langs     :: Array Lang
-             , onSearch  :: GT.AsyncTaskWithType -> Effect Unit
+             , onSearch  :: GAT.Task -> Effect Unit
              , search    :: T.Box Search
              , session   :: Session
              )

src/Gargantext/Components/Forest/Tree/Node/Action/Search/SearchField.purs

@@ -13,6 +13,7 @@ import Data.Tuple (Tuple(..))
 import Effect (Effect)
 import Effect.Aff (launchAff_)
 import Effect.Class (liftEffect)
+import Gargantext.AsyncTasks as GAT
 import Gargantext.Components.Forest.Tree.Node.Action.Search.Frame (searchIframes)
 import Gargantext.Components.Forest.Tree.Node.Action.Search.Types (DataField(..), Database(..), IMT_org(..), Org(..), SearchQuery(..), allOrgs, dataFields, defaultSearchQuery, doc, performSearch, datafield2database, Search, dbFromInputValue, dbToInputValue)
 import Gargantext.Components.GraphQL.Endpoints (getIMTSchools, getUser)
@@ -53,7 +54,7 @@ type Props =
   , errors    :: T.Box (Array FrontendError)
   , langs     :: Array Lang
   -- State hook for a search, how we get data in and out
-  , onSearch  :: GT.AsyncTaskWithType -> Effect Unit
+  , onSearch  :: GAT.Task -> Effect Unit
   , search    :: T.Box Search
   , session   :: Session
   )
@@ -645,7 +646,7 @@ searchInputCpt = here.component "searchInput" cpt
 
 type SubmitButtonProps =
   ( errors    :: T.Box (Array FrontendError)
-  , onSearch  :: GT.AsyncTaskWithType -> Effect Unit
+  , onSearch  :: GAT.Task -> Effect Unit
   , search    :: T.Box Search
   , selection :: T.Box ListSelection.Selection
   , session   :: Session
@@ -677,13 +678,13 @@ submitButtonComponent = here.component "submitButton" cpt
 
 type TriggerSearch =
   ( errors    :: T.Box (Array FrontendError)
-  , onSearch  :: GT.AsyncTaskWithType -> Effect Unit
+  , onSearch  :: GAT.Task -> Effect Unit
   , search    :: T.Box Search
   , selection :: T.Box ListSelection.Selection
   , session   :: Session
   )
 
-triggerSearch :: { onSearch :: (GT.AsyncTaskWithType -> Effect Unit)
+triggerSearch :: { onSearch :: (GAT.Task -> Effect Unit)
                  , errors :: T.Box (Array FrontendError)
                  , session :: Session
                  , selection :: ListSelection.Selection

src/Gargantext/Components/Forest/Tree/Node/Action/Search/Types.purs

@@ -12,6 +12,7 @@ import Data.Set (Set)
 import Data.Set as Set
 import Data.Tuple (Tuple)
 import Data.Tuple.Nested ((/\))
+import Gargantext.AsyncTasks as GAT
 import Gargantext.Components.GraphQL.IMT as GQLIMT
 import Gargantext.Components.Lang (Lang)
 import Gargantext.Components.ListSelection.Types as ListSelection
@@ -339,9 +340,10 @@ defaultSearchQuery = SearchQuery
   , selection    : ListSelection.NoList -- MyListsFirst
   }
 
-performSearch :: Session -> Int -> SearchQuery -> AffRESTError GT.AsyncTaskWithType
+performSearch :: Session -> Int -> SearchQuery -> AffRESTError GAT.Task
 performSearch session nodeId q = do
-  eTask :: Either RESTError GT.AsyncTask <- post session p q
-  pure $ (\task -> GT.AsyncTaskWithType { task, typ: GT.Query }) <$> eTask
+  post session p q
+  -- eTask :: Either RESTError GT.AsyncTask <- post session p q
+  -- pure $ (\task -> GAT.Task { task, typ: GT.Query }) <$> eTask
   where
     p = GR.NodeAPI GT.Corpus (Just nodeId) $ GT.asyncTaskTypePath GT.Query

src/Gargantext/Components/Forest/Tree/Node/Action/Types.purs

@@ -2,6 +2,7 @@ module Gargantext.Components.Forest.Tree.Node.Action.Types where
 
 import Data.Generic.Rep (class Generic)
 import Data.Maybe (Maybe)
+import Gargantext.AsyncTasks as GAT
 import Gargantext.Components.Forest.Tree.Node.Action.Contact.Types (AddContactParams)
 import Gargantext.Components.Forest.Tree.Node.Action.Update.Types (UpdateNodeParams)
 import Gargantext.Components.Forest.Tree.Node.Action.Upload.Types (FileFormat, FileType, UploadFileBlob)
@@ -15,7 +16,7 @@ data Action = AddNode     String GT.NodeType
             | DeleteNode  GT.NodeType
             | RenameNode  String
             | UpdateNode  UpdateNodeParams
-            | DoSearch    GT.AsyncTaskWithType
+            | DoSearch    GAT.Task
             | UploadFile  GT.NodeType FileType FileFormat Lang (Maybe String) String Selection
             | UploadArbitraryFile FileFormat (Maybe String) UploadFileBlob
             | UploadFrameCalc Lang Selection

src/Gargantext/Components/Notifications.purs

@@ -56,11 +56,11 @@ callNotification (State { callbacks }) n = do
   -- here.log2 "[callTopic] callbacks" (HM.values callbacks)
   -- here.log2 "[callTopic] topicCallbacks" (HM.values topicCallbacks)
 
-  let topic = notificationTopic n
-  
-  _ <- for (HM.values $ topicCallbacks topic) $ \cb -> do
-    cb n
-  pure unit
+  let topics = notificationTopics n
+
+  void $ for topics $ \topic -> do
+    void $ for (HM.values $ topicCallbacks topic) $ \cb -> do
+      cb n
   where
     topicCallbacks :: Topic -> CallbacksHM
     topicCallbacks topic = fromMaybe HM.empty $ HM.lookup topic callbacks

src/Gargantext/Components/Notifications/Types.purs

@@ -86,6 +86,7 @@ instance JSON.WriteForeign WSRequest where
 
 data Notification =
     NUpdateJobProgress GT.AsyncTaskID GT.AsyncProgress
+  | NUpdateWorkerProgress GT.WorkerTask GT.AsyncTaskLog
   | NUpdateTree NodeId
 derive instance Generic Notification _
 instance JSON.ReadForeign Notification where
@@ -95,14 +96,26 @@ instance JSON.ReadForeign Notification where
       "update_job_progress" -> do
         { j_id, job_status } <- JSON.readImpl f :: F.F { j_id :: GT.AsyncTaskID, job_status :: GT.AsyncProgress }
         pure $ NUpdateJobProgress j_id job_status
+      "update_worker_progress" -> do
+        { job_info, job_log } <- JSON.readImpl f :: F.F { job_info :: GT.WorkerTask, job_log :: GT.AsyncTaskLog }
+        pure $ NUpdateWorkerProgress job_info job_log
       "update_tree" -> do
         { node_id } <- JSON.readImpl f :: F.F { node_id :: NodeId }
         pure $ NUpdateTree node_id
       s -> F.fail $ F.ErrorAtProperty "type" $ F.ForeignError $ "unkown type: " <> s
 
-notificationTopic :: Notification -> Topic
-notificationTopic (NUpdateJobProgress taskId _) = UpdateJobProgress taskId
-notificationTopic (NUpdateTree nodeId) = UpdateTree nodeId
+notificationTopics :: Notification -> Array Topic
+notificationTopics (NUpdateJobProgress taskId _) = [ UpdateJobProgress taskId ]
+notificationTopics (NUpdateWorkerProgress workerTask@(GT.WorkerTask { node_id }) _) =
+  [ UpdateWorkerProgress workerTask ] <> updateTree
+  where
+    -- when receiving a worker progress notification, we are also
+    -- interested in 'update tree' subscriptions, because there might
+    -- be a new job that we didn't subscribe to
+    updateTree = case node_id of
+      Nothing -> []
+      Just nId -> [ UpdateTree nId ]
+notificationTopics (NUpdateTree nodeId) = [ UpdateTree nodeId ]
 
 
 type Callback = Notification -> Effect Unit

src/Gargantext/Config/Utils.purs

@@ -10,7 +10,7 @@ import Effect (Effect)
 import Effect.Aff (Aff)
 import Effect.Class (liftEffect)
 import Gargantext.Config.REST (RESTError, logRESTError)
-import Gargantext.Types (AsyncEvent(..), AsyncProgress(..), AsyncTaskLog(..), AsyncTaskStatus(..), FrontendError(..))
+import Gargantext.Types (AsyncEvent(..), AsyncProgress(..), AsyncTaskLog(..), AsyncTaskStatus(..), FrontendError(..), asyncTaskLogEventsErrorMessage)
 import Gargantext.Utils.Reactix as R2
 import Toestand as T
 
@@ -46,6 +46,11 @@ handleErrorInAsyncProgress _ _ = pure unit
 concatErrors :: AsyncProgress -> String
 concatErrors (AsyncProgress { error, log }) = foldl eventsErrorMessage (fromMaybe "" error) log
   where
-    eventsErrorMessage acc (AsyncTaskLog { events }) = (foldl eventErrorMessage "" events) <> "\n" <> acc
-    eventErrorMessage acc (AsyncEvent { level: "ERROR", message }) = message <> "\n" <> acc
-    eventErrorMessage acc _ = acc
+    eventsErrorMessage acc atl = asyncTaskLogEventsErrorMessage atl <> "\n" <> acc
+
+
+handleErrorInAsyncTaskLog :: T.Box (Array FrontendError)
+                          -> AsyncTaskLog
+                          -> Effect Unit
+handleErrorInAsyncTaskLog errors atl = do
+  T.modify_ (A.cons $ FStringError { error: asyncTaskLogEventsErrorMessage atl }) errors

src/Gargantext/Context/Progress.purs

@@ -5,6 +5,7 @@ module Gargantext.Context.Progress
   ) where
 
 import Data.Either (Either(..))
+import Data.Foldable (foldl)
 import Data.Maybe (Maybe(..))
 import Data.Tuple.Nested ((/\))
 import Effect (Effect)
@@ -16,7 +17,7 @@ import Gargantext.Components.App.Store as AppStore
 import Gargantext.Components.Forest.Tree.Node.Tools.ProgressBar (QueryProgressData, queryProgress)
 import Gargantext.Components.Notifications as Notifications
 import Gargantext.Components.Notifications.Types as NT
-import Gargantext.Config.Utils (handleErrorInAsyncProgress, handleRESTError)
+import Gargantext.Config.Utils (handleErrorInAsyncTaskLog, handleRESTError)
 import Gargantext.Hooks.FirstEffect (useFirstEffect')
 import Gargantext.Prelude
 import Gargantext.Sessions (Session)
@@ -29,7 +30,7 @@ import Toestand as T
 
 
 type AsyncProps =
-  ( asyncTask :: GT.AsyncTaskWithType
+  ( asyncTask :: GAT.Task
   , nodeId    :: GT.ID
   , onFinish  :: Unit -> Effect Unit
   , session   :: Session
@@ -60,80 +61,99 @@ asyncProgressCpt = R2.hereComponent here "asyncProgress" hCpt where
     interval <- T.useBox 1000
 
     -- Methods
-    let
-      -- TODO Manage somehow to get the whole job status sent here via
-      -- websockets, then we can remove the 'Maybe'
-      fetchJobProgress :: Effect Unit
-      fetchJobProgress = launchAff_ do
-        let rdata = (RX.pick props :: Record QueryProgressData)
-
-        eAsyncProgress <- queryProgress rdata
-        -- liftEffect $ here.log2 "[progress] received asyncProgress" eAsyncProgress
-
-        -- exponential backoff in case of errors
-        -- liftEffect $ do
-        --   case eAsyncProgress of
-        --     Left _  -> T.modify_ (_ * 2) interval
-        --     Right _ -> T.write_ 1000 interval
-        --   interval' <- T.read interval
-        --   resetInterval intervalIdRef (Just interval') exec
-
-        -- Handle removal of task in case of 500 error (e.g. server
-        -- was restarted and task id is not found anymore).
-        -- Error logging will be done below, in handleRESTError
-        case eAsyncProgress of
-          Right _ -> pure unit
-          Left err -> do
-            liftEffect $ do
-              resetInterval intervalIdRef Nothing (pure unit)
-              GAT.finish props.nodeId props.asyncTask tasks
-
-        handleRESTError hp errors eAsyncProgress onProgress
-
-      -- TODO Ideally we should use this function
-      -- onProgress jobProgress = do
-      --   launchAff_ $ onProgress jobProgress
+    -- let
+    --   -- TODO Manage somehow to get the whole job status sent here via
+    --   -- websockets, then we can remove the 'Maybe'
+    --   fetchJobProgress :: Effect Unit
+    --   fetchJobProgress = launchAff_ do
+    --     let rdata = (RX.pick props :: Record QueryProgressData)
+
+    --     eAsyncProgress <- queryProgress rdata
+    --     -- liftEffect $ here.log2 "[progress] received asyncProgress" eAsyncProgress
+
+    --     -- exponential backoff in case of errors
+    --     -- liftEffect $ do
+    --     --   case eAsyncProgress of
+    --     --     Left _  -> T.modify_ (_ * 2) interval
+    --     --     Right _ -> T.write_ 1000 interval
+    --     --   interval' <- T.read interval
+    --     --   resetInterval intervalIdRef (Just interval') exec
+
+    --     -- Handle removal of task in case of 500 error (e.g. server
+    --     -- was restarted and task id is not found anymore).
+    --     -- Error logging will be done below, in handleRESTError
+    --     case eAsyncProgress of
+    --       Right _ -> pure unit
+    --       Left err -> do
+    --         liftEffect $ do
+    --           resetInterval intervalIdRef Nothing (pure unit)
+    --           GAT.finish props.nodeId props.asyncTask tasks
+
+    --     handleRESTError hp errors eAsyncProgress onProgress
+
+    --   -- TODO Ideally we should use this function
+    --   -- onProgress jobProgress = do
+    --   --   launchAff_ $ onProgress jobProgress
         
 
-      onProgress :: AsyncProgress -> Aff Unit
-      onProgress value@(GT.AsyncProgress { status }) = liftEffect do
+    let
+      -- onProgress :: AsyncProgress -> Aff Unit
+      -- onProgress value@(GT.AsyncProgress { status }) = liftEffect do
+      onProgress :: GT.AsyncTaskLog -> Aff Unit
+      onProgress atl@(GT.AsyncTaskLog log) = liftEffect do
 
-        T.write_ (min 100.0 $ GT.progressPercent value) progressBox
+        T.write_ (min 100.0 $ GT.asyncTaskLogPercent atl) progressBox
 
-        if (status == GT.IsFinished) ||
-           (status == GT.IsKilled) ||
-           (status == GT.IsFailure)
+        here.log "[onProgress] TODO: Implement status killed"
+        if GT.asyncTaskLogIsFinished atl ||
+           GT.asyncTaskLogIsError atl
         then do
-          -- resetInterval intervalIdRef Nothing exec
-          -- case R.readRef intervalIdRef of
-          --   Nothing  -> R.nothing
-          --   Just iid -> clearInterval iid
-          handleErrorInAsyncProgress errors value
-          resetInterval intervalIdRef Nothing (pure unit)
+          handleErrorInAsyncTaskLog errors atl
+          -- resetInterval intervalIdRef Nothing (pure unit)
           onFinish unit
         else
           R.nothing
 
+        -- if (status == GT.IsFinished) ||
+        --    (status == GT.IsKilled) ||
+        --    (status == GT.IsFailure)
+        -- then do
+        --   -- resetInterval intervalIdRef Nothing exec
+        --   -- case R.readRef intervalIdRef of
+        --   --   Nothing  -> R.nothing
+        --   --   Just iid -> clearInterval iid
+        --   handleErrorInAsyncProgress errors value
+        --   resetInterval intervalIdRef Nothing (pure unit)
+        --   onFinish unit
+        -- else
+        --   R.nothing
+
     useFirstEffect' $ do
-      let (GT.AsyncTaskWithType { task: GT.AsyncTask { id: taskId } }) = props.asyncTask
+      -- let (GT.AsyncTaskWithType { task: GT.AsyncTask { id: taskId } }) = props.asyncTask
+      let (GT.WorkerTask { message_id }) = props.asyncTask
       let cb n = do
             -- here.log2 "callback! for job update" taskId
             case n of
-              NT.NUpdateJobProgress _jId jobProgress -> launchAff_ $ onProgress jobProgress
+              NT.NUpdateWorkerProgress _wt jobProgress -> launchAff_ $ onProgress jobProgress
               _ -> pure unit
 
-            resetInterval intervalIdRef (Just defaultJobPollInterval) fetchJobProgress
+            -- resetInterval intervalIdRef (Just defaultJobPollInterval) fetchJobProgress
 
             -- The modal window has some problems closing when we refresh too early. This is a HACK
             -- void $ setTimeout 400 $ T2.reload reload
-      let action = NT.InsertCallback (NT.UpdateJobProgress taskId) ("task-" <> show taskId) cb
       ws <- T.read wsNotification
+      let action = NT.InsertCallback (NT.UpdateWorkerProgress props.asyncTask) ("task-" <> show message_id) cb
       Notifications.performAction ws action
+
+      -- ws <- T.read wsNotification
+      -- New-style jobs
       -- let action = NT.InsertCallback (NT.UpdateWorkerProgress $ GT.WorkerTask { message_id: taskId }) ("worker-job-" <> show taskId) cb
-      let action = NT.InsertCallback (NT.UpdateJobProgress taskId) ("job-" <> taskId) cb
-      ws <- T.read wsNotification
-      Notifications.performAction ws action
-      fetchJobProgress
+      -- Notifications.performAction ws action
+      -- Old-style jobs (remove in the future)
+      -- let action = NT.InsertCallback (NT.UpdateJobProgress taskId) ("job-" <> taskId) cb
+      -- Notifications.performAction ws action
+
+      -- fetchJobProgress
  
     -- Hooks
     -- useFirstEffect' do
@@ -145,8 +165,8 @@ asyncProgressCpt = R2.hereComponent here "asyncProgress" hCpt where
     -- itself, notify us when a job finished. Hence, we are forced to
     -- poll for job still. However, we will keep canceling the timer
     -- unless there is no progress report for some time.
-    useFirstEffect' $ do
-      resetInterval intervalIdRef (Just defaultJobPollInterval) fetchJobProgress
+    -- useFirstEffect' $ do
+    --   resetInterval intervalIdRef (Just defaultJobPollInterval) fetchJobProgress
 
     -- Render
     pure $

src/Gargantext/Core/NgramsTable/Functions.purs

@@ -584,7 +584,9 @@ chartsAfterSync :: forall props discard.
 chartsAfterSync path'@{ nodeId } errors tasks _ = do
   eTask <- postNgramsChartsAsync path'
   handleRESTError (R2.herePrefix here "[chartsAfterSync]") errors eTask $ \task -> liftEffect $ do
-    GAT.insert nodeId task tasks
+    -- GAT.insert nodeId task tasks
+    here.log "[chartsAfterSync] TODO: IMPLEMENT ME!"
+
 
 postNgramsChartsAsync :: forall s. CoreParams s -> AffRESTError AsyncTaskWithType
 postNgramsChartsAsync { listIds, nodeId, session, tabType } = do

src/Gargantext/Types.purs

@@ -5,6 +5,7 @@ import Gargantext.Prelude
 import Data.Argonaut as Argonaut
 import Data.Array as A
 import Data.Eq.Generic (genericEq)
+import Data.Foldable (foldl)
 import Data.Generic.Rep (class Generic)
 import Data.Int (toNumber)
 import Data.Maybe (Maybe(..), maybe)
@@ -815,6 +816,10 @@ derive instance Generic AsyncEvent _
 derive instance Newtype AsyncEvent _
 derive newtype instance JSON.ReadForeign AsyncEvent
 
+asyncEventErrorMessage :: AsyncEvent -> Maybe String
+asyncEventErrorMessage (AsyncEvent { level: "ERROR", message }) = Just message
+asyncEventErrorMessage _ = Nothing
+
 newtype AsyncTaskLog = AsyncTaskLog
   { events :: Array AsyncEvent
   , failed :: Int
@@ -825,23 +830,44 @@ derive instance Generic AsyncTaskLog _
 derive instance Newtype AsyncTaskLog _
 derive newtype instance JSON.ReadForeign AsyncTaskLog
 
+asyncTaskLogEventsErrorMessage :: AsyncTaskLog -> String
+asyncTaskLogEventsErrorMessage (AsyncTaskLog { events }) =
+  foldl eventErrorMessage' "" events
+  where
+    eventErrorMessage' acc ae = (case asyncEventErrorMessage ae of
+                                    Nothing -> ""
+                                    Just e' -> e' <> "\n") <> acc
+asyncTaskLogPercent :: AsyncTaskLog -> Number
+asyncTaskLogPercent (AsyncTaskLog { failed, remaining, succeeded }) = 100.0 * nom / denom
+  where
+    nom = toNumber $ failed + succeeded
+    denom = toNumber $ failed + succeeded + remaining
+
+asyncTaskLogIsFinished :: AsyncTaskLog -> Boolean
+asyncTaskLogIsFinished (AsyncTaskLog { remaining }) = remaining == 0
+
+asyncTaskLogIsError :: AsyncTaskLog -> Boolean
+asyncTaskLogIsError atl@(AsyncTaskLog { events }) =
+  asyncTaskLogIsFinished atl &&
+  (A.length $ A.filter (\(AsyncEvent { level }) -> level == "ERROR") events) > 0
+
 progressPercent :: AsyncProgress -> Number
 progressPercent (AsyncProgress { log }) = perc
   where
     perc = case A.head log of
       Nothing -> 0.0
-      Just (AsyncTaskLog {failed, remaining, succeeded}) -> 100.0*nom/denom
-        where
-          nom = toNumber $ failed + succeeded
-          denom = toNumber $ failed + succeeded + remaining
+      Just atl -> asyncTaskLogPercent atl
 
 
 -- New type tasks (async workers)
-newtype WorkerTask = WorkerTask { message_id :: Number }
+newtype WorkerTask = WorkerTask { message_id :: Number
+                                , node_id    :: Maybe NodeID }
 derive instance Generic WorkerTask _
 derive instance Newtype WorkerTask _
 instance Eq WorkerTask where
   eq = genericEq
+instance Ord WorkerTask where
+  compare = genericCompare
 instance Show WorkerTask where
   show = genericShow
 derive newtype instance JSON.ReadForeign WorkerTask