module Router.Nats ( RouteDirective (..) , routeMessage ) where import Control.Monad (when) import Control.Monad.IO.Class (liftIO) import qualified Data.ByteString as BS import Lib.Logger (LogLevel (..), MonadLogger (..)) import Parser.API ( ParsedMessage (ParsedErr, ParsedInfo, ParsedMessageTooLarge, ParsedMsg, ParsedOk, ParsedPing, ParsedPong) ) import Queue.API ( QueueItem (QueueConnectionScoped, QueueItem) ) import State.Store ( ClientState , ServerErrorEnqueueResult (..) , enqueue , notifyServerError , resolveNextPing , runClient , setServerInfo , updateLogContextFromInfo ) import State.Types ( ClientExitReason (ExitInboundMessageTooLarge, ExitServerError) , serverErrorFromProtocol , serverErrorKind ) import Subscription.Store ( DispatchResult (DispatchDropped, DispatchMissing, DispatchQueued) , SubscriptionStore , dispatchMessage ) import qualified Types.Err as Err import qualified Types.Msg as Msg import Types.Pong (Pong (..)) data RouteDirective = RouteContinue | RouteReconnect | RouteExit ClientExitReason deriving (Eq, Show) routeMessage :: ClientState -> SubscriptionStore -> ParsedMessage -> IO RouteDirective routeMessage state store parsed = runClient state $ case parsed of ParsedMsg msg -> do logMessage Debug ( "routing MSG: subject=" ++ show (Msg.subject msg) ++ " sid=" ++ show (Msg.sid msg) ++ " payload_bytes=" ++ show (maybe 0 BS.length (Msg.payload msg)) ++ " header_count=" ++ show (maybe 0 length (Msg.headers msg)) ) result <- liftIO $ dispatchMessage store msg case result of DispatchQueued -> pure RouteContinue DispatchDropped reportSlowConsumer -> do if reportSlowConsumer then logMessage Error "slow consumer: global pending delivery limit reached" else logMessage Debug "dropping delivery while client remains a slow consumer" pure RouteContinue DispatchMissing -> do logMessage Error ("callback missing for SID: " ++ show (Msg.sid msg)) pure RouteContinue ParsedMessageTooLarge actual maximumSize -> do logMessage Error ( "inbound message size " ++ show actual ++ " exceeds client limit " ++ show maximumSize ) pure (RouteExit (ExitInboundMessageTooLarge actual maximumSize)) ParsedInfo info -> do logMessage Debug "routing INFO" liftIO $ setServerInfo state info liftIO $ updateLogContextFromInfo state info pure RouteContinue ParsedPing _ -> do logMessage Debug "routing PING" result <- liftIO $ enqueue state (QueueConnectionScoped (QueueItem Pong)) pure $ case result of Left _ -> RouteReconnect Right () -> RouteContinue ParsedPong _ -> do logMessage Debug "routing PONG" liftIO $ resolveNextPing state pure RouteContinue ParsedOk okMsg -> do logMessage Debug ("routing OK: " ++ show okMsg) pure RouteContinue ParsedErr err -> do let serverError = serverErrorFromProtocol err kind = serverErrorKind serverError logMessage Debug ("routing ERR: kind=" ++ show kind) enqueueResult <- liftIO $ notifyServerError state serverError let accepted = enqueueResult == ServerErrorQueued when (enqueueResult == ServerErrorDroppedReport) $ logMessage Warn "server error handler queue full; dropping event" if Err.isFatal err then do logMessage Error ("fatal server error: kind=" ++ show kind) pure (RouteExit (ExitServerError serverError)) else if accepted then do logMessage Warn ("server error: kind=" ++ show kind) pure RouteContinue else pure RouteContinue