{-|
Module      : Test.API.Worker
Description : Basic tests for the async worker
Copyright   : (c) CNRS, 2017-Present
License     : AGPL + CECILL v3
Maintainer  : team@gargantext.org
Stability   : experimental
Portability : POSIX

-}

{-# OPTIONS_GHC -Wno-orphans  #-}

{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TemplateHaskell #-}


module Test.API.Worker (
    tests
  ) where

import Control.Concurrent.Async (withAsync, forConcurrently_)
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))
import Network.WebSockets qualified as WS
import Prelude
import System.Timeout qualified as Timeout
import Test.API.Setup (SpecContext(..), withTestDBAndPort)
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
import Test.HUnit (assertBool)



tests :: Spec
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))

      withAsync (setupWsThread log_cfg topic tchan port) $ \_a -> do
        _ <- sendJobWithCfg cfg Ping

        mTimeout <- Timeout.timeout (5 * 1_000_000) $ do
          md <- atomically $ readTChan tchan

          md `shouldBe` Just DT.NPing

        mTimeout `shouldSatisfy` isJust

    describe "concurrency" $ do

      -- This test checks that two concurrent threads can both subscribe
      -- to the same topic and get notified.
      it "handles concurrent threads" $ \(SpecContext testEnv port _app _) -> do
        let cfg = test_config testEnv
        let log_cfg = (test_config testEnv) ^. gc_logging

        let topic = DT.Ping
        let competingThreads = 3

        forConcurrently_ [ 1 .. competingThreads ] $ \(tid :: Int) -> do
          tchan <- newTChanIO :: IO (TChan (Maybe DT.Notification))

          withAsync (setupWsThread log_cfg topic tchan port) $ \_a -> do
            _ <- sendJobWithCfg cfg Ping

            mTimeout <- Timeout.timeout (5 * 1_000_000) $ do
              md <- atomically $ readTChan tchan

              md `shouldBe` Just DT.NPing

            assertBool ("Competing Thread " <> show tid <> " didn't receive a value.") (isJust mTimeout)

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
