refactor(logging): Add more debug logs for dispatcher/workers

They can be enabled with `GGTX_LOG_LEVEL` during tests.

src/Gargantext/Core/Notifications/Dispatcher.hs

@@ -14,6 +14,8 @@ https://dev.sub.gargantext.org/#/share/Notes/187918
 
 -}
 
+{-# LANGUAGE TemplateHaskell #-}
+
 module Gargantext.Core.Notifications.Dispatcher (
     Dispatcher -- opaque
   , withDispatcher
@@ -34,11 +36,11 @@ import Gargantext.Core.Notifications.CentralExchange.Types qualified as CETypes
 import Gargantext.Core.Notifications.Dispatcher.Types
 import Gargantext.Core.Worker.Types (JobInfo(..))
 import Gargantext.Prelude
-import Gargantext.System.Logging (LogLevel(..), withLogger, logMsg)
 import Nanomsg (Pull(..), bind, recv, withSocket)
 import Network.WebSockets qualified as WS
 import StmContainers.Set qualified as SSet
 import Gargantext.Core.Config
+import Gargantext.System.Logging
 
 {-
 
@@ -92,20 +94,18 @@ dispatcherListener config subscriptions = do
   where
     NotificationsConfig { _nc_dispatcher_bind } = config ^. gc_notifications_config
     log_cfg = config ^. gc_logging
-    worker tChan throttleTChan = do
-      -- tId <- myThreadId
+    worker tChan throttleTChan = withLogger log_cfg $ \ioL -> do
+      tId <- myThreadId
 
       forever $ do
         r <- atomically $ TChan.readTChan tChan
-        -- putText $ "[" <> show tId <> "] received a message: " <> decodeUtf8 r
+        $(logLoc) ioL DEBUG $ "[" <> show tId <> "] received a message: " <> decodeUtf8 r
 
         case Aeson.decode (BSL.fromStrict r) of
           Nothing ->
-            withLogger log_cfg $ \ioL ->
               logMsg ioL DEBUG "[dispatcher_listener] unknown message from central exchange"
           Just ceMessage -> do
-            withLogger log_cfg $ \ioL ->
-              logMsg ioL DEBUG $ "[dispatcher_listener] received " <> show ceMessage
+            logMsg ioL DEBUG $ "[dispatcher_listener] received " <> show ceMessage
             -- subs <- atomically $ readTVar subscriptions
             filteredSubs <- atomically $ do
               let subs' = UnfoldlM.filter (pure . ceMessageSubPred ceMessage) $ SSet.unfoldlM subscriptions

src/Gargantext/Core/Notifications/Dispatcher/Types.hs

@@ -11,11 +11,11 @@ https://gitlab.iscpif.fr/gargantext/haskell-gargantext/issues/341
 
 Docs:
 https://dev.sub.gargantext.org/#/share/Notes/187918
-    
+
 -}
 
 {-# OPTIONS_GHC -fno-warn-unused-matches -fno-warn-unused-imports #-}
-    
+
 module Gargantext.Core.Notifications.Dispatcher.Types where
 
 import Codec.Binary.UTF8.String qualified as CBUTF8
@@ -120,7 +120,7 @@ instance ToJSON Topic where
 --     pure $ MJobLog jl
 
 
-    
+
 data ConnectedUser =
     CUUser UserId
   | CUPublic
@@ -128,7 +128,7 @@ data ConnectedUser =
 instance Hashable ConnectedUser where
   hashWithSalt salt (CUUser userId) = hashWithSalt salt ("cuuser" :: Text, userId)
   hashWithSalt salt CUPublic = hashWithSalt salt ("cupublic" :: Text)
-  
+
 newtype WSKeyConnection = WSKeyConnection (ByteString, WS.Connection)
 instance Hashable WSKeyConnection where
   hashWithSalt salt (WSKeyConnection (key, _conn)) = hashWithSalt salt key
@@ -142,7 +142,7 @@ wsKey :: WSKeyConnection -> ByteString
 wsKey (WSKeyConnection (key, _conn)) = key
 wsConn :: WSKeyConnection -> WS.Connection
 wsConn (WSKeyConnection (_key, conn)) = conn
-    
+
 data Subscription =
   Subscription {
       s_connected_user    :: ConnectedUser
@@ -158,7 +158,7 @@ subKey sub = wsKey $ s_ws_key_connection sub
 
 type Token = Text
 
-    
+
 {-
 We accept requests for subscription/unsubscription via websocket.
 
@@ -200,7 +200,7 @@ instance ToJSON WSRequest where
   toJSON (WSAuthorize token) = Aeson.object [ "request" .= ( "authorize" :: Text )
                                             , "token" .= token ]
   toJSON WSDeauthorize = Aeson.object [ "request" .= ( "deauthorize" :: Text ) ]
-    
+
 class HasDispatcher env dispatcher where
   hasDispatcher :: Getter env dispatcher
 

src/Gargantext/Core/Notifications/Dispatcher/WebSocket.hs

@@ -100,6 +100,7 @@ wsLoop log_cfg jwtS subscriptions ws = flip finally disconnect $ do
   where
     wsLoop' user ioLogger = do
       dm <- WS.receiveDataMessage (wsConn ws)
+      logMsg ioLogger DEBUG $ "[wsLoop'] handling new message.."
 
       newUser <- case dm of
         WS.Text dm' _ -> do
@@ -113,8 +114,8 @@ wsLoop log_cfg jwtS subscriptions ws = flip finally disconnect $ do
               let sub = Subscription { s_connected_user = user
                                      , s_ws_key_connection = ws
                                      , s_topic = topic }
-              _ss <- insertSubscription subscriptions sub
-              -- putText $ "[wsLoop] subscriptions: " <> show (showSub <$> ss)
+              insertSubscription subscriptions sub
+              logMsg ioLogger DEBUG $ "[wsLoop] added subscription: " <> show sub
               return user
             Just (WSUnsubscribe topic) -> do
               logMsg ioLogger DEBUG $ "[wsLoop'] unsubscribe topic " <> show topic

test/Test/API/Worker.hs

@@ -12,6 +12,7 @@ Portability : POSIX
 {-# OPTIONS_GHC -Wno-orphans  #-}
 
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TemplateHaskell #-}
 
 
 module Test.API.Worker (
@@ -20,9 +21,11 @@ module Test.API.Worker (
 
 import Control.Concurrent.Async (withAsync)
 import Control.Concurrent.STM.TChan
+import Control.Lens
 import Control.Monad.STM (atomically)
 import Data.Aeson qualified as Aeson
 import Data.Maybe (isJust)
+import Gargantext.Core.Config
 import Gargantext.Core.Notifications.Dispatcher.Types qualified as DT
 import Gargantext.Core.Worker.Jobs (sendJobWithCfg)
 import Gargantext.Core.Worker.Jobs.Types (Job(Ping))
@@ -34,6 +37,9 @@ import Test.Database.Types (test_config)
 import Test.Hspec
 import Test.Instances ()
 import Test.Utils.Notifications
+import Gargantext.System.Logging
+import qualified Data.Text.Encoding as TE
+import qualified Data.ByteString as BL
 
 
 
@@ -42,18 +48,12 @@ tests = sequential $ aroundAll withTestDBAndPort $ do
   describe "Worker" $ do
     it "simple Ping job works" $ \(SpecContext testEnv port _app _) -> do
       let cfg = test_config testEnv
+      let log_cfg = (test_config testEnv) ^. gc_logging
 
       let topic = DT.Ping
       tchan <- newTChanIO :: IO (TChan (Maybe DT.Notification))
-      -- setup a websocket connection
-      let wsConnect =
-            withWSConnection ("127.0.0.1", port) $ \conn -> do
-              WS.sendTextData conn $ Aeson.encode (DT.WSSubscribe topic)
-              d <- WS.receiveData conn
-              let dec = Aeson.decode d :: Maybe DT.Notification
-              atomically $ writeTChan tchan dec
 
-      withAsync wsConnect $ \_a -> do
+      withAsync (setupWsThread log_cfg topic tchan port) $ \_a -> do
         _ <- sendJobWithCfg cfg Ping
 
         mTimeout <- Timeout.timeout (5 * 1_000_000) $ do
@@ -62,3 +62,14 @@ tests = sequential $ aroundAll withTestDBAndPort $ do
           md `shouldBe` Just DT.NPing
 
         mTimeout `shouldSatisfy` isJust
+
+setupWsThread :: LogConfig -> DT.Topic -> TChan (Maybe DT.Notification) -> Int -> IO ()
+setupWsThread log_cfg topic tchan port = withLogger log_cfg $ \ioL -> do
+  withWSConnection ("127.0.0.1", port) $ \conn -> do
+    let payload = Aeson.encode (DT.WSSubscribe topic)
+    $(logLoc) ioL DEBUG $ "Sending payload: " <> (TE.decodeUtf8 $ BL.toStrict $ payload)
+    WS.sendTextData conn $ Aeson.encode (DT.WSSubscribe topic)
+    d <- WS.receiveData conn
+    $(logLoc) ioL DEBUG $ "Received: " <> (TE.decodeUtf8 $ BL.toStrict d)
+    let dec = Aeson.decode d :: Maybe DT.Notification
+    atomically $ writeTChan tchan dec