{-|
Module      : Gargantext.Core.Notifications.Dispatcher.Types
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

-}

{-# 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
import Control.Concurrent.Async qualified as Async
import Control.Lens (Getter, view)
import Data.Aeson ((.:), (.=))
import Data.Aeson qualified as Aeson
import Data.Aeson.Types (prependFailure, typeMismatch)
import Data.ByteString.Char8 qualified as C
import Data.ByteString.Lazy qualified as BSL
import Data.List (nubBy)
import DeferredFolds.UnfoldlM qualified as UnfoldlM
import Data.UUID.V4 as UUID
import Gargantext.API.Admin.Auth.Types (AuthenticatedUser(_auth_user_id))
import Gargantext.API.Admin.Orchestrator.Types (JobLog)
import Gargantext.API.Prelude (IsGargServer)
import Gargantext.Core.Notifications.CentralExchange.Types qualified as CETypes
import Gargantext.Core.Types (NodeId, UserId)
import Gargantext.Core.Worker.Types (JobInfo)
import Gargantext.Prelude
import GHC.Conc (TVar, newTVarIO, readTVar, writeTVar)
import Nanomsg
import Network.WebSockets qualified as WS
import Prelude qualified
import Protolude.Base (Show(showsPrec))
import Servant
-- import Servant.API.NamedRoutes ((:-))
import Servant.API.WebSocket qualified as WS
import Servant.Auth.Server (verifyJWT)
import Servant.Server.Generic (AsServer, AsServerT)
import StmContainers.Set as SSet



-- | A topic is sent, when a client wants to subscribe to specific
-- | types of notifications
data Topic =
  -- | New, worker version for updating job state
    UpdateWorkerProgress JobInfo
  -- | Given parent node id, trigger update of the node and its
  --   children (e.g. list is automatically created in a corpus)
  | UpdateTree NodeId
  | Ping
  deriving (Eq, Ord)
instance Prelude.Show Topic where
  show (UpdateWorkerProgress ji) = "UpdateWorkerProgress " <> show ji
  show (UpdateTree nodeId) = "UpdateTree " <> show nodeId
  show Ping = "Ping"
instance Hashable Topic where
  hashWithSalt salt (UpdateWorkerProgress ji) = hashWithSalt salt ("update-worker-progress" :: Text, Aeson.encode ji)
  hashWithSalt salt (UpdateTree nodeId) = hashWithSalt salt ("update-tree" :: Text, nodeId)
  hashWithSalt salt Ping = hashWithSalt salt ("ping" :: Text)
instance FromJSON Topic where
  parseJSON = Aeson.withObject "Topic" $ \o -> do
    type_ <- o .: "type"
    case type_ of
      "update_worker_progress" -> do
        ji <- o .: "ji"
        pure $ UpdateWorkerProgress ji
      "update_tree" -> do
        node_id <- o .: "node_id"
        pure $ UpdateTree node_id
      "ping" -> pure Ping
      s -> prependFailure "parsing type failed, " (typeMismatch "type" s)
instance ToJSON Topic where
  toJSON (UpdateWorkerProgress ji) = Aeson.object [
      "type" .= ("update_worker_progress" :: Text)
    , "ji" .= ji
    ]
  toJSON (UpdateTree node_id) = Aeson.object [
      "type" .= ("update_tree" :: Text)
    , "node_id" .= node_id
    ]
  toJSON Ping = Aeson.object [ "type" .= ("ping" :: Text) ]

-- | A job status message
-- newtype MJobStatus = MJobStatus (JobStatus 'Safe JobLog)
-- instance Prelude.Show MJobStatus where
--   show (MJobStatus js) = "MJobStatus " <> show (CBUTF8.decode $ BSL.unpack $ Aeson.encode js)
-- instance ToJSON MJobStatus where
--   toJSON (MJobStatus js) = Aeson.object [
--       "type" .= toJSON ("MJobLog" :: Text)
--     , "job_status" .= toJSON js
--     ]
-- instance FromJSON MJobStatus where
--   parseJSON = Aeson.withObject "MJobStatus" $ \o -> do
--     js <- o .: "job_status"
--     pure $ MJobStatus js

-- | A job progress message
-- newtype MJobLog = MJobLog JobLog
-- instance Prelude.Show MJobLog where
--   show (MJobLog jl) = "MJobLog " <> show jl
-- instance ToJSON MJobLog where
--   toJSON (MJobLog jl) = Aeson.object [
--       "type" .= toJSON ("MJobLog" :: Text)
--     , "job_log" .= toJSON jl
--     ]
-- instance FromJSON MJobLog where
--   parseJSON = Aeson.withObject "MJobLog" $ \o -> do
--     jl <- o .: "job_log"
--     pure $ MJobLog jl



data ConnectedUser =
    CUUser UserId
  | CUPublic
  deriving (Eq, Show)
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
instance Eq WSKeyConnection where
  (==) (WSKeyConnection (key1, _conn1)) (WSKeyConnection (key2, _conn2)) = key1 == key2
instance Show WSKeyConnection where
  showsPrec d (WSKeyConnection (key, _conn)) = showsPrec d $ "WSKeyConnection " <> key
showWSKeyConnection :: WSKeyConnection -> Text
showWSKeyConnection ws = "WSKeyConnection " <> show (wsKey ws)
wsKey :: WSKeyConnection -> ByteString
wsKey (WSKeyConnection (key, _conn)) = key
wsConn :: WSKeyConnection -> WS.Connection
wsConn (WSKeyConnection (_key, conn)) = conn

data Subscription =
  Subscription {
      s_connected_user    :: ConnectedUser
    , s_ws_key_connection :: WSKeyConnection
    , s_topic             :: Topic }
    deriving (Eq, Show)
instance Hashable Subscription where
  hashWithSalt salt (Subscription { .. }) =
    hashWithSalt salt ( s_connected_user, s_ws_key_connection, s_topic )
subKey :: Subscription -> ByteString
subKey sub = wsKey $ s_ws_key_connection sub


type Token = Text


{-
We accept requests for subscription/unsubscription via websocket.

We could instead handle 1 websocket connection per every topic
subscription (e.g. parse headers in WS.PendingConnection. However, WS
by default can handle 65k concurrent connections. With multiple users
having multiple components open, we could exhaust that limit quickly.

Hence, we architect this to have 1 websocket connection per web
browser.
-}
data WSRequest =
    WSSubscribe Topic
  | WSUnsubscribe Topic
  | WSAuthorize Token
  | WSDeauthorize
  deriving (Eq, Show)
instance FromJSON WSRequest where
  parseJSON = Aeson.withObject "WSRequest" $ \o -> do
    request <- o .: "request"
    case request of
      "subscribe" -> do
        topic <- o .: "topic"
        pure $ WSSubscribe topic
      "unsubscribe" -> do
        topic <- o .: "topic"
        pure $ WSUnsubscribe topic
      "authorize" -> do
         token <- o .: "token"
         pure $ WSAuthorize token
      "deauthorize" -> pure $ WSDeauthorize
      s -> prependFailure "parsing request type failed, " (typeMismatch "request" s)
-- | For tests mainly
instance ToJSON WSRequest where
  toJSON (WSSubscribe topic) = Aeson.object [ "request" .= ( "subscribe":: Text )
                                            , "topic" .= topic ]
  toJSON (WSUnsubscribe topic) = Aeson.object [ "request" .= ( "unsubscribe" :: Text )
                                              , "topic" .= topic ]
  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


-- | A notification is sent to clients who subscribed to specific topics
data Notification =
    NUpdateWorkerProgress JobInfo JobLog
  | NUpdateTree NodeId
  | NWorkerJobStarted NodeId JobInfo
  | NWorkerJobFinished NodeId JobInfo
  | NPing
instance Prelude.Show Notification where
  -- show (NUpdateWorkerProgress jobInfo nodeId mJobLog) = "NUpdateWorkerProgress " <> show jobInfo <> ", " <> show nodeId <> ", " <> show mJobLog
  show (NUpdateWorkerProgress jobInfo mJobLog) = "NUpdateWorkerProgress " <> show jobInfo <> ", " <> show mJobLog
  show (NUpdateTree nodeId) = "NUpdateTree " <> show nodeId
  show (NWorkerJobStarted nodeId ji) = "NWorkerJobStarted " <> show nodeId <> ", " <> show ji
  show (NWorkerJobFinished nodeId ji) = "NWorkerJobFinished " <> show nodeId <> ", " <> show ji
  show NPing = "NPing"
instance ToJSON Notification where
  -- toJSON (NUpdateWorkerProgress jobInfo nodeId mJobLog) = Aeson.object [
  toJSON (NUpdateWorkerProgress jobInfo mJobLog) = Aeson.object [
      "type" .= ("update_worker_progress" :: Text)
    , "job_info" .= toJSON jobInfo
    , "job_log"  .= toJSON mJobLog
    -- , "node_id"  .= toJSON nodeId
    ]
  toJSON (NUpdateTree nodeId) = Aeson.object [
      "type" .= ("update_tree" :: Text)
    , "node_id" .= toJSON nodeId
    ]
  toJSON (NWorkerJobStarted nodeId ji) = Aeson.object [
      "type" .= ("worker_job_started" :: Text)
    , "node_id" .= toJSON nodeId
    , "ji" .= toJSON ji
    ]
  toJSON (NWorkerJobFinished nodeId ji) = Aeson.object [
      "type" .= ("worker_job_finished" :: Text)
    , "node_id" .= toJSON nodeId
    , "ji" .= toJSON ji
    ]
  toJSON NPing = Aeson.object [ "type" .= ("ping" :: Text) ]
-- We don't need to decode notifications, this is for tests only
instance FromJSON Notification where
  parseJSON = Aeson.withObject "Notification" $ \o -> do
    t <- o .: "type"
    case t of
      "update_worker_progress" -> do
        jobInfo <- o .: "job_info"
        mJobLog <- o .: "job_log"
        -- nodeId <- o .: "node_id"
        -- pure $ NUpdateWorkerProgress jobInfo nodeId mJobLog
        pure $ NUpdateWorkerProgress jobInfo mJobLog
      "update_tree" -> do
        nodeId <- o .: "node_id"
        pure $ NUpdateTree nodeId
      "worker_job_started" -> do
        nodeId <- o .: "node_id"
        ji <- o .: "ji"
        pure $ NWorkerJobStarted nodeId ji
      "worker_job_finished" -> do
        nodeId <- o .: "node_id"
        ji <- o .: "ji"
        pure $ NWorkerJobFinished nodeId ji
      "ping" -> pure NPing
      s -> prependFailure "parsing type failed, " (typeMismatch "type" s)
