Request.hs

{-# LANGUAGE OverloadedStrings #-}

module JetStream.Protocol.Request
  ( requestMsg
  , requestJSON
  , requestJSONWithHeaders
  , requestUnit
  , decodeJetStreamResponse
  ) where

import qualified API                        as Nats
import           Control.Concurrent.STM
import           Control.Monad              (void)
import           Data.Aeson
import qualified Data.Aeson                 as Aeson
import           Data.Aeson.Types           (Parser, parseEither)
import qualified Data.ByteString            as BS
import qualified Data.ByteString.Lazy       as LBS
import           JetStream.Error
import           JetStream.Options          (JetStreamContext (..))
import           JetStream.Protocol.Headers (statusError)
import           JetStream.Types            (Headers, Payload, Subject)
import           System.Timeout             (timeout)

requestMsg :: JetStreamContext -> Subject -> Maybe Payload -> Headers -> IO (Either JetStreamError Nats.MsgView)
requestMsg ctx subject payload headers = do
  replyBox <- newEmptyTMVarIO
  let publishOptions =
        maybe [] (\body -> [Nats.withPayload body]) payload
          ++ [Nats.withReplyCallback (atomically . putTMVar replyBox)]
          ++ [Nats.withHeaders headers | not (null headers)]
  Nats.publish (contextClient ctx) subject publishOptions
  result <- timeout (contextRequestTimeoutMicros ctx) (atomically (readTMVar replyBox))
  case result of
    Nothing ->
      pure (Left JetStreamTimeout)
    Just Nothing ->
      pure (Left JetStreamNoReply)
    Just (Just msg) ->
      pure (Right msg)

requestJSON :: FromJSON a => JetStreamContext -> Subject -> Maybe Value -> IO (Either JetStreamError a)
requestJSON ctx subject body =
  requestJSONWithHeaders ctx subject body []

requestJSONWithHeaders :: FromJSON a => JetStreamContext -> Subject -> Maybe Value -> Headers -> IO (Either JetStreamError a)
requestJSONWithHeaders ctx subject body headers = do
  msgResult <- requestMsg ctx subject (strictEncode <$> body) headers
  pure $ do
    msg <- msgResult
    case statusError msg of
      Just err ->
        Left err
      Nothing ->
        case Nats.payload msg of
          Nothing ->
            Left JetStreamNoReply
          Just payload ->
            decodeJetStreamResponse payload

requestUnit :: JetStreamContext -> Subject -> Maybe Value -> IO (Either JetStreamError ())
requestUnit ctx subject body = do
  response <- requestJSON ctx subject body
  pure (void (response :: Either JetStreamError SuccessResponse))

decodeJetStreamResponse :: FromJSON a => BS.ByteString -> Either JetStreamError a
decodeJetStreamResponse payload =
  case Aeson.eitherDecode (LBS.fromStrict payload) :: Either String Value of
    Left err ->
      Left (JetStreamDecodeError err)
    Right value ->
      case parseEither parseApiResult value of
        Left err ->
          Left (JetStreamDecodeError err)
        Right result ->
          result

parseApiResult :: FromJSON a => Value -> Parser (Either JetStreamError a)
parseApiResult value =
  case value of
    Object obj -> do
      apiError <- obj .:? "error"
      case apiError of
        Just err ->
          pure (Left (JetStreamApiFailure err))
        Nothing ->
          Right <$> parseJSON value
    _ ->
      Right <$> parseJSON value

strictEncode :: Value -> BS.ByteString
strictEncode =
  LBS.toStrict . Aeson.encode

newtype SuccessResponse = SuccessResponse Bool
  deriving (Eq, Show)

instance FromJSON SuccessResponse where
  parseJSON = withObject "SuccessResponse" $ \obj ->
    SuccessResponse <$> obj .: "success"