{-|
Module      : Test.API.Notifications
Description : Tests for the notification mechanism (central exchange, dispatcher, websockets)
Copyright   : (c) CNRS, 2017-Present
License     : AGPL + CECILL v3
Maintainer  : team@gargantext.org
Stability   : experimental
Portability : POSIX

-}

{-# OPTIONS_GHC -Wno-orphans  #-}

{-# LANGUAGE QuasiQuotes      #-}
{-# LANGUAGE ScopedTypeVariables #-}


module Test.API.Notifications (
    tests
  ) where

import Control.Concurrent.STM.TChan
import Control.Concurrent.STM.TSem (newTSem, signalTSem, TSem)
import Control.Concurrent (threadDelay)
import Control.Lens ((^.))
import Control.Monad (void)
import Control.Monad.STM (atomically)
import Data.Aeson qualified as Aeson
import Data.ByteString qualified as BS
import Data.Text.Encoding qualified as TE
import Data.Text qualified as T
import Fmt ((+|), (|+))
import Gargantext.API.Admin.Auth.Types (AuthResponse, authRes_token, authRes_tree_id)
import Gargantext.Core.Config (gc_logging, LogConfig)
import Gargantext.Core.Notifications.CentralExchange qualified as CE
import Gargantext.Core.Notifications.CentralExchange.Types qualified as CET
import Gargantext.Core.Notifications.Dispatcher.Types qualified as DT
import Gargantext.Core.Types.Individu (GargPassword(..))
import Gargantext.System.Logging (withLogger)
import Network.WebSockets qualified as WS
import Prelude
import System.Timeout qualified as Timeout
import Test.API.Prelude (newCorpusForUser)
import Test.API.Routes (mkUrl)
import Test.API.Setup (SpecContext(..), dbEnvSetup, withTestDBAndPort)
import Test.Database.Types (test_config)
import Test.Hspec
import Test.Hspec.Wai.Internal (withApplication)
import Test.Instances ()
import Test.Utils (waitForTChanValue, waitForTSem)
import Test.Utils.Notifications (withAsyncWSConnection)
import Test.Utils (protected, withValidLoginA)
import Text.RawString.QQ (r)




tests :: Spec
tests = sequential $ around withTestDBAndPort $ do
  describe "Notifications" $ do
    it "ping WS notification works" $ \(SpecContext testEnv port _app _) -> do
      let cfg     = test_config testEnv
      let log_cfg = (test_config testEnv) ^. gc_logging
      -- withLogger () $ \ioL -> do
      --   logMsg ioL DEBUG $ "[ping WS notification works] nc: " <> show nc

      let topic = DT.Ping
      -- This semaphore is used to inform the main thread that the WS
      -- client has subscribed. I think it's better to use async
      -- locking mechanisms than blindly call 'threadDelay'.
      wsTSem <- atomically $ newTSem 0
      tchan <- newTChanIO :: IO (TChan (Maybe DT.Notification))

      withAsyncWSConnection ("127.0.0.1", port) (wsConnection log_cfg topic wsTSem tchan) $ \_a -> do
        -- wait for ws process to inform us about topic subscription
        waitForTSem wsTSem 500

        threadDelay 300_000
        CE.notify cfg $ CET.Ping

        -- the ping value that should come from the notification
        waitForTChanValue tchan (Just DT.NPing) 1_000

    it "ping WS unsubscribe works" $ \(SpecContext testEnv port _app _) -> do
      let cfg     = test_config testEnv
      let log_cfg = (test_config testEnv) ^. gc_logging
      let topic = DT.Ping

      -- Setup a WS client connection. Subscribe to a topic and
      -- confirm the notification works. Then unsubscribe from it, and
      -- check that a new notification didn't arrive.

      wsTSem <- atomically $ newTSem 0
      tchan <- newTChanIO :: IO (TChan (Maybe DT.Notification))

      -- setup a websocket connection
      let wsConnect conn = withLogger log_cfg $ \_ioL -> do
            -- logMsg ioL DEBUG $ "[wsConnect] subscribing topic: " <> show topic
            WS.sendTextData conn $ Aeson.encode (DT.WSSubscribe topic)
            -- inform the test process that we sent the subscription request
            atomically $ signalTSem wsTSem

            -- logMsg ioL DEBUG $ "[wsConnect] waiting for notification"
            d <- WS.receiveData conn
            -- logMsg ioL DEBUG $ "[wsConnect] received " <> show d
            let dec = Aeson.decode d :: Maybe DT.Notification
            atomically $ writeTChan tchan dec

            -- now ubsubscribe from a topic and make sure nothing arrives
            WS.sendTextData conn $ Aeson.encode (DT.WSUnsubscribe topic)
            -- Signal that we finished unsubscribing
            atomically $ signalTSem wsTSem
            mTimeout <- Timeout.timeout (200_000) $ do
              -- NOTE This shouldn't happen now, we will test the tchan
              d' <- WS.receiveData conn
              let dec' = Aeson.decode d' :: Maybe DT.Notification
              atomically $ writeTChan tchan dec'
            case mTimeout of
              -- It should have timed out
              Nothing -> atomically $ writeTChan tchan Nothing
              -- | write something incorrect so the test will fail
              Just _ -> atomically $ writeTChan tchan (Just DT.NPing)

      withAsyncWSConnection ("127.0.0.1", port) wsConnect $ \_a -> do
        -- wait for ws process to inform us about topic subscription
        waitForTSem wsTSem 500

        threadDelay 300_000
        CE.notify cfg $ CET.Ping

        -- the ping value that should come from the notification
        waitForTChanValue tchan (Just DT.NPing) 1_000

        -- wait for lock from ws (it should have unsubscribed by now)
        waitForTSem wsTSem 500
        -- send the notification (which the client shouldn't receive)
        CE.notify cfg $ CET.Ping
        -- wait for the value
        waitForTChanValue tchan Nothing 1_000

    describe "Update tree notifications" $ do
      it "simple WS notification works" $ \(SpecContext testEnv port _app _) -> do
        let topic = DT.UpdateTree 0
        wsTSem <- atomically $ newTSem 0  -- initially locked
        tchan <- newTChanIO :: IO (TChan (Maybe DT.Notification))

        withAsyncWSConnection ("127.0.0.1", port) (wsConnection (test_config testEnv ^. gc_logging) topic wsTSem tchan) $ \_a -> do
          waitForTSem wsTSem 500

          let nodeId = 0
          CE.notify (test_config testEnv) $ CET.UpdateTreeFirstLevel nodeId

          waitForTChanValue tchan (Just $ DT.NUpdateTree nodeId) 1_000

      it "WS notification on node creation works" $ \ctx@(SpecContext _testEnv port app _) -> do
        checkNotification ctx $ \authRes -> do
          let token = authRes ^. authRes_token
          let treeId = authRes ^. authRes_tree_id
          let query = [r| {"pn_name": "test", "pn_typename": "NodeCorpus"} |]
          void $ withApplication app $ do
            protected token "POST" (mkUrl port $ "/node/" +| treeId |+ "") query

      it "WS notification on node deletion works" $ \ctx@(SpecContext testEnv port app _) -> do
        checkNotification ctx $ \authRes -> do
          let token = authRes ^. authRes_token
          cId <- newCorpusForUser testEnv "alice"

          void $ withApplication app $ do
            protected token "DELETE" (mkUrl port $ "/node/" +| cId |+ "") ""

      it "WS notification on node rename works" $ \ctx@(SpecContext testEnv port app _) -> do
        checkNotification ctx $ \authRes -> do
          let token = authRes ^. authRes_token
          cId <- newCorpusForUser testEnv "alice"

          void $ withApplication app $ do
            let query = [r| {"name": "newName"} |]
            protected token "PUT" (mkUrl port $ "/node/" +| cId |+ "/rename") query

      it "WS notification on node move works" $ \ctx@(SpecContext testEnv port app _) -> do
        checkNotification ctx $ \authRes -> do
          let token = authRes ^. authRes_token
          cId <- newCorpusForUser testEnv "alice"
          cId2 <- newCorpusForUser testEnv "alice"

          void $ withApplication app $ do
            let query = BS.fromStrict $ TE.encodeUtf8 $ "[" <> (T.pack $ show cId2) <> "]"
            protected token "PUT" (mkUrl port $ "/node/" +| cId |+ "/move/" +| cId2 |+ "" ) query



-- | Given spec context and an action, call that action to perform
-- some node tree update, and check that there was a notification
-- about this tree update.
checkNotification :: SpecContext a
                  -> (AuthResponse -> IO ())
                  -> IO ()
checkNotification ctx@(SpecContext testEnv port _app _) act = do
  _ <- dbEnvSetup ctx

  withValidLoginA port "alice" (GargPassword "alice") $ \_clientEnv authRes -> do
    -- Subscribe to user tree notifications
    let treeId = authRes ^. authRes_tree_id
    let topic = DT.UpdateTree treeId

    wsTSem <- atomically $ newTSem 0  -- initially locked
    tchan <- newTChanIO :: IO (TChan (Maybe DT.Notification))

    withAsyncWSConnection ("127.0.0.1", port) (wsConnection log_cfg topic wsTSem tchan) $ \_a -> do
      waitForTSem wsTSem 500

      act authRes

      -- Wait /up to/ 5 seconds for the notification value. This makes running the tests
      -- a bit less flaky on CI, which might be slower to process the incoming notifications.
      waitForTChanValue tchan (Just $ DT.NUpdateTree treeId) 5_000
  where
    log_cfg = (test_config testEnv) ^. gc_logging


wsConnection :: LogConfig
             -> DT.Topic
             -> TSem
             -> TChan (Maybe DT.Notification)
             -> WS.Connection
             -> IO ()
wsConnection log_cfg topic wsTSem tchan conn = withLogger log_cfg $ \_ioL -> do
  -- logMsg ioL DEBUG $ "[wsConnect] subscribing topic: " <> show topic
  WS.sendTextData conn $ Aeson.encode (DT.WSSubscribe topic)
  -- inform the test process that we sent the subscription request
  atomically $ signalTSem wsTSem

  -- logMsg ioL DEBUG $ "[wsConnect] waiting for notification"
  d <- WS.receiveData conn
  -- logMsg ioL DEBUG $ "[wsConnect] received " <> show d
  let dec = Aeson.decode d :: Maybe DT.Notification
  atomically $ writeTChan tchan dec

