{-|
Module      : Gargantext.Core.AsyncUpdates.CentralExchange
Description : Central exchange (asynchronous notifications)
Copyright   : (c) CNRS, 2017-Present
License     : AGPL + CECILL v3
Maintainer  : team@gargantext.org
Stability   : experimental
Portability : POSIX

https://gitlab.iscpif.fr/gargantext/haskell-gargantext/issues/341

Docs:
https://dev.sub.gargantext.org/#/share/Notes/187918
    
-}

module Gargantext.Core.AsyncUpdates.CentralExchange (
    gServer
  , notify
  ) where

import Control.Concurrent.Async qualified as Async
import Control.Concurrent.STM.TChan qualified as TChan
import Data.Aeson qualified as Aeson
import Data.ByteString.Lazy qualified as BSL
import Data.Text qualified as T
import Data.Text.Encoding qualified as TE
import Gargantext.Core.AsyncUpdates.CentralExchange.Types
import Gargantext.Core.Config.Types (NotificationsConfig(..))
import Gargantext.Prelude
import Gargantext.System.Logging (LogLevel(..), withLogger, logMsg)
import Nanomsg (Pull(..), Push(..), bind, connect, recv, sendNonblocking, withSocket)
    
{-

Central exchange is a service, which gathers messages from various
places and informs the Dispatcher (which will then inform users about
various events).

The primary goal is to be able to read as many messages as possible
and then send them to the Dispatcher. Although nanomsg does some
message buffering, we don't want these messages to pile up, especially
with many users having updates.
    
-}

gServer :: NotificationsConfig -> IO ()
gServer (NotificationsConfig { .. }) = do
  withSocket Pull $ \s -> do
    withSocket Push $ \s_dispatcher -> do
      _ <- bind s $ T.unpack _nc_central_exchange_bind
      _ <- connect s_dispatcher $ T.unpack _nc_dispatcher_connect

      tChan <- TChan.newTChanIO

      -- | We have 2 threads: one that listens for nanomsg messages
      -- | and puts them on the 'tChan' and the second one that reads
      -- | the 'tChan' and calls Dispatcher accordingly. This is to
      -- | make reading nanomsg as fast as possible.
      void $ Async.concurrently (worker s_dispatcher tChan) $ do
        withLogger () $ \ioLogger -> do
          forever $ do
            -- putText "[central_exchange] receiving"
            r <- recv s
            logMsg ioLogger INFO $ "[central_exchange] received: " <> show r
            -- C.putStrLn $ "[central_exchange] " <> r
            atomically $ TChan.writeTChan tChan r
  where
    worker s_dispatcher tChan = do
      withLogger () $ \ioLogger -> do
        forever $ do
          r <- atomically $ TChan.readTChan tChan
          case Aeson.decode (BSL.fromStrict r) of
            Just _ujp@(UpdateJobProgress _s) -> do
              -- logMsg ioLogger DEBUG $ "[central_exchange] " <> show ujp
              -- send the same message that we received
              void $ sendNonblocking s_dispatcher r
            Just (UpdateTreeFirstLevel node_id) -> do
              logMsg ioLogger INFO $ "[central_exchange] update tree: " <> show node_id
              -- putText $ "[central_exchange] sending that to the dispatcher: " <> show node_id
              -- To make this more robust, use withAsync so we don't
              -- block the main thread (send is blocking)
              
              -- NOTE: If we're flooded with messages, and send is
              -- slow, we might be spawning many threads...
    
              -- NOTE: Currently we just forward the message that we
              -- got. So in theory central exchange isn't needed (we
              -- could ping dispatcher directly). However, I think
              -- it's better to have this as a separate
              -- component. Currently I built this inside
              -- gargantext-server but maybe it can be a separate
              -- process, independent of the server.
              -- send the same message that we received
              void $ sendNonblocking s_dispatcher r
            _ -> logMsg ioLogger DEBUG $ "[central_exchange] unknown message"
      

notify :: NotificationsConfig -> CEMessage -> IO ()
notify (NotificationsConfig { _nc_central_exchange_connect }) ceMessage = do
  Async.withAsync (pure ()) $ \_ -> do
    withSocket Push $ \s -> do
      _ <- connect s $ T.unpack _nc_central_exchange_connect
      let str = Aeson.encode ceMessage
      withLogger () $ \ioLogger ->
        logMsg ioLogger DEBUG $ "[central_exchange] sending: " <> (T.unpack $ TE.decodeUtf8 $ BSL.toStrict str)
      void $ sendNonblocking s $ BSL.toStrict str
