{-|
Module      : Gargantext.Core.AsyncUpdates.Dispatcher
Description : Dispatcher (handles websocket connections, accepts message from central exchange)
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
    
-}

{-# LANGUAGE TypeOperators #-}
    
module Gargantext.Core.AsyncUpdates.Dispatcher (
    Dispatcher -- opaque
  , newDispatcher
  , terminateDispatcher

  -- * Querying a dispatcher
  , dispatcherSubscriptions
  ) where

import Control.Concurrent.Async qualified as Async
import Control.Concurrent.STM.TChan qualified as TChan
import Control.Concurrent.Throttle (throttle)
import Data.Aeson qualified as Aeson
import Data.ByteString.Lazy qualified as BSL
import Data.Text qualified as T
import DeferredFolds.UnfoldlM qualified as UnfoldlM
import Gargantext.Core.AsyncUpdates.CentralExchange.Types qualified as CETypes
import Gargantext.Core.AsyncUpdates.Dispatcher.Types
import Gargantext.Core.Config.Types (NotificationsConfig(..))
import Gargantext.Prelude
import Gargantext.System.Logging (LogLevel(DEBUG), withLogger, logMsg)
import Nanomsg (Pull(..), bind, recv, withSocket)
import Network.WebSockets qualified as WS
import Servant.Job.Types (JobStatus(_job_id))
import StmContainers.Set qualified as SSet

{-

Dispatcher is a service, which provides couple of functionalities:
- handles WebSocket connections and manages them
- accepts messages from central exchange
- dispatches these messages to connected users
    
-}

data Dispatcher =
 Dispatcher { d_subscriptions :: SSet.Set Subscription
            -- , d_ws_server     :: WSAPI AsServer
            , d_ce_listener   :: ThreadId
            }

terminateDispatcher :: Dispatcher -> IO ()
terminateDispatcher = killThread . d_ce_listener

dispatcherSubscriptions :: Dispatcher -> SSet.Set Subscription
dispatcherSubscriptions = d_subscriptions

newDispatcher :: NotificationsConfig -> IO Dispatcher
newDispatcher nc = do
  subscriptions <- SSet.newIO

  -- let server = wsServer authSettings subscriptions

  d_ce_listener <- forkIO (dispatcherListener nc subscriptions)

  pure $ Dispatcher { d_subscriptions = subscriptions
                    -- , d_ws_server     = server
                    , d_ce_listener   = d_ce_listener }


    
-- | This is a nanomsg socket listener. We want to read the messages
-- | as fast as possible and then process them gradually in a separate
-- | thread.
dispatcherListener :: NotificationsConfig -> SSet.Set Subscription -> IO ()
dispatcherListener (NotificationsConfig { _nc_dispatcher_bind }) subscriptions = do
  withSocket Pull $ \s -> do
    _ <- bind s $ T.unpack _nc_dispatcher_bind

    tChan <- TChan.newTChanIO

    throttleTChan <- TChan.newTChanIO

    -- NOTE I'm not sure that we need more than 1 worker here, but in
    -- theory, the worker can perform things like user authentication,
    -- DB queries etc so it can be slow sometimes.
    Async.withAsync (throttle 500_000 throttleTChan sendDataMessageThrottled) $ \_ -> do
      void $ Async.concurrently (Async.replicateConcurrently 5 $ worker tChan throttleTChan) $ do
        forever $ do
          -- putText "[dispatcher_listener] receiving"
          r <- recv s
          -- C.putStrLn $ "[dispatcher_listener] " <> r
          atomically $ TChan.writeTChan tChan r
  where
    worker tChan throttleTChan = do
      -- tId <- myThreadId
      
      forever $ do
        r <- atomically $ TChan.readTChan tChan
        -- putText $ "[" <> show tId <> "] received a message: " <> decodeUtf8 r
        
        case Aeson.decode (BSL.fromStrict r) of
          Nothing -> withLogger () $ \ioL ->
                       logMsg ioL DEBUG "[dispatcher_listener] unknown message from central exchange"
          Just ceMessage -> do
            -- putText $ "[dispatcher_listener] received message: " <> show ceMessage
            -- subs <- atomically $ readTVar subscriptions
            filteredSubs <- atomically $ do
              let subs' = UnfoldlM.filter (pure . ceMessageSubPred ceMessage) $ SSet.unfoldlM subscriptions
              UnfoldlM.foldlM' (\acc sub -> pure $ acc <> [sub]) [] subs'
            -- NOTE This isn't safe: we atomically fetch subscriptions,
            -- then send notifications one by one. In the meantime, a
            -- subscription could end or new ones could appear (but is
            -- this really a problem? I new subscription comes up, then
            -- probably they already fetch new tree anyways, and if old
            -- one drops in the meantime, it won't listen to what we
            -- send...)
            -- let filteredSubs = filterCEMessageSubs ceMessage subs
            mapM_ (sendNotification throttleTChan ceMessage) filteredSubs

-- | When processing tasks such as Flow, we can generate quite a few
-- notifications in a short time. We want to limit this with throttle
-- tchan.
sendNotification :: TChan.TChan ((ByteString, Topic), (WS.Connection, WS.DataMessage))
                 -> CETypes.CEMessage
                 -> Subscription
                 -> IO ()
sendNotification throttleTChan ceMessage sub = do
  let ws = s_ws_key_connection sub
  let topic = s_topic sub
  notification <-
        case ceMessage of
          CETypes.UpdateJobProgress jobStatus -> do
            pure $ Notification topic (MJobProgress jobStatus)
          CETypes.UpdateTreeFirstLevel _nodeId -> pure $ Notification topic MEmpty

  let id' = (wsKey ws, topic)
  atomically $ TChan.writeTChan throttleTChan (id', (wsConn ws, WS.Text (Aeson.encode notification) Nothing))

sendDataMessageThrottled :: (WS.Connection, WS.DataMessage) -> IO ()
sendDataMessageThrottled (conn, msg) =
  WS.sendDataMessage conn msg


-- Custom filtering of list of Subscriptions based on
-- CETypes.CEMessage.
-- For example, we can add CEMessage.Broadcast to propagate a
-- notification to all connections.
_filterCEMessageSubs :: CETypes.CEMessage -> [Subscription] -> [Subscription]
_filterCEMessageSubs ceMessage subscriptions = filter (ceMessageSubPred ceMessage) subscriptions

ceMessageSubPred :: CETypes.CEMessage -> Subscription -> Bool
ceMessageSubPred (CETypes.UpdateJobProgress js) (Subscription { s_topic }) =
  s_topic == (UpdateJobProgress $ _job_id js)
ceMessageSubPred (CETypes.UpdateTreeFirstLevel node_id) (Subscription { s_topic }) =
  s_topic == UpdateTree node_id
