{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE UndecidableInstances #-}

module Hydra.API.WSServer where

import Hydra.Prelude hiding (TVar, filter, readTVar, seq)

import Cardano.Binary (serialize')
import Conduit (ConduitT, ResourceT, mapM_C, runConduitRes, (.|))
import Control.Concurrent.STM (TChan, dupTChan, readTChan)
import Control.Concurrent.STM qualified as STM
import Data.Aeson qualified as Aeson
import Data.ByteString.Char8 qualified as BS8
import Data.ByteString.Lazy qualified as LBS
import Data.Conduit.Combinators (filter)
import Data.Version (showVersion)
import Hydra.API.APIServerLog (APIServerLog (..))
import Hydra.API.ClientInput (ClientInput (SafeClose))
import Hydra.API.Projection (Projection (..))
import Hydra.API.ServerOutput (
  ApiEncoding (..),
  ClientMessage,
  Greetings (..),
  HeadStatus (..),
  InvalidInput (..),
  NetworkInfo,
  ServerOutputConfig (..),
  TimedServerOutput (..),
  WithAddressedTx (..),
  WithUTxO (..),
  getSnapshotUtxo,
  handleUtxoInclusionTyped,
  headStatus,
  me,
  prepareServerOutput,
  snapshotUtxo,
 )
import Hydra.API.ServerOutputFilter (
  ServerOutputFilter (..),
 )
import Hydra.API.WireFormat (decodeWire, describeWire)
import Hydra.Chain (Chain (..))
import Hydra.Chain.ChainState (IsChainState)
import Hydra.HeadLogic (ClosedState (ClosedState, readyToFanoutSent), HeadState, OpenState (..), PartialFanoutState (..), StateChanged)
import Hydra.HeadLogic.State qualified as HeadState
import Hydra.Logging (Tracer, traceWith)
import Hydra.NetworkVersions qualified as NetworkVersions
import Hydra.Node.Environment (Environment (..))
import Hydra.Node.State (ChainPointTime (..), NodeState (..), syncedStatus)
import Hydra.Tx (HeadId, Party)
import Network.HTTP.Types.URI (Query, parseQuery)
import Network.WebSockets (
  Connection,
  PendingConnection (pendingRequest),
  RequestHead (..),
  acceptRequest,
  receiveData,
  sendBinaryData,
  sendTextData,
  withPingThread,
 )

-- | Per-connection codec: resolves the negotiated wire encoding and the
-- snapshot-utxo display policy once, so message handling needs no dispatch.
data WsCodec tx = WsCodec
  { forall tx. WsCodec tx -> TimedServerOutput tx -> IO ()
sendOutput :: TimedServerOutput tx -> IO ()
  , forall tx. WsCodec tx -> ClientMessage tx -> IO ()
sendClientMessage :: ClientMessage tx -> IO ()
  , forall tx. WsCodec tx -> Greetings tx -> IO ()
sendGreetings :: Greetings tx -> IO ()
  , forall tx. WsCodec tx -> InvalidInput -> IO ()
sendInvalidInput :: InvalidInput -> IO ()
  , forall tx.
WsCodec tx -> ByteString -> Either String (ClientInput tx)
decodeInput :: LBS.ByteString -> Either String (ClientInput tx)
  , forall tx. WsCodec tx -> ByteString -> Text
describeInput :: LBS.ByteString -> Text
  }

-- | Resolve the negotiated encoding into a 'WsCodec'. This is the only place
-- deciding how messages go onto (and come off) the wire: JSON as text frames,
-- CBOR as binary frames.
--
-- NOTE: Inputs are decoded per the negotiated encoding, never the frame type:
-- some JSON clients (e.g. the TUI) send binary frames containing JSON.
mkWsCodec :: IsChainState tx => ServerOutputConfig -> Connection -> WsCodec tx
mkWsCodec :: forall tx.
IsChainState tx =>
ServerOutputConfig -> Connection -> WsCodec tx
mkWsCodec ServerOutputConfig
config Connection
con =
  case ServerOutputConfig
config.encoding of
    ApiEncoding
JsonEncoding ->
      WsCodec
        { $sel:sendOutput:WsCodec :: TimedServerOutput tx -> IO ()
sendOutput = Connection -> ByteString -> IO ()
forall a. WebSocketsData a => Connection -> a -> IO ()
sendTextData Connection
con (ByteString -> IO ())
-> (TimedServerOutput tx -> ByteString)
-> TimedServerOutput tx
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ServerOutputConfig -> TimedServerOutput tx -> ByteString
forall tx.
IsChainState tx =>
ServerOutputConfig -> TimedServerOutput tx -> ByteString
prepareServerOutput ServerOutputConfig
config
        , -- NOTE: 'ClientMessage' has no top-level snapshot, so the
          -- snapshot-utxo filter does not apply to it.
          $sel:sendClientMessage:WsCodec :: ClientMessage tx -> IO ()
sendClientMessage = ClientMessage tx -> IO ()
forall a. ToJSON a => a -> IO ()
sendPlainJson
        , $sel:sendGreetings:WsCodec :: Greetings tx -> IO ()
sendGreetings = Greetings tx -> IO ()
forall a. ToJSON a => a -> IO ()
sendPlainJson
        , $sel:sendInvalidInput:WsCodec :: InvalidInput -> IO ()
sendInvalidInput = InvalidInput -> IO ()
forall a. ToJSON a => a -> IO ()
sendPlainJson
        , $sel:decodeInput:WsCodec :: ByteString -> Either String (ClientInput tx)
decodeInput = ApiEncoding -> ByteString -> Either String (ClientInput tx)
forall a.
(FromJSON a, FromCBOR a) =>
ApiEncoding -> ByteString -> Either String a
decodeWire ApiEncoding
JsonEncoding
        , $sel:describeInput:WsCodec :: ByteString -> Text
describeInput = ApiEncoding -> ByteString -> Text
describeWire ApiEncoding
JsonEncoding
        }
    ApiEncoding
CborEncoding ->
      WsCodec
        { $sel:sendOutput:WsCodec :: TimedServerOutput tx -> IO ()
sendOutput = Connection -> ByteString -> IO ()
forall a. WebSocketsData a => Connection -> a -> IO ()
sendBinaryData Connection
con (ByteString -> IO ())
-> (TimedServerOutput tx -> ByteString)
-> TimedServerOutput tx
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TimedServerOutput tx -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' (TimedServerOutput tx -> ByteString)
-> (TimedServerOutput tx -> TimedServerOutput tx)
-> TimedServerOutput tx
-> ByteString
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ServerOutputConfig -> TimedServerOutput tx -> TimedServerOutput tx
forall tx.
IsTx tx =>
ServerOutputConfig -> TimedServerOutput tx -> TimedServerOutput tx
handleUtxoInclusionTyped ServerOutputConfig
config
        , -- NOTE: 'ClientMessage' has no top-level snapshot, so the
          -- snapshot-utxo filter does not apply to it.
          $sel:sendClientMessage:WsCodec :: ClientMessage tx -> IO ()
sendClientMessage = ClientMessage tx -> IO ()
forall a. ToCBOR a => a -> IO ()
sendPlainCbor
        , $sel:sendGreetings:WsCodec :: Greetings tx -> IO ()
sendGreetings = Greetings tx -> IO ()
forall a. ToCBOR a => a -> IO ()
sendPlainCbor
        , $sel:sendInvalidInput:WsCodec :: InvalidInput -> IO ()
sendInvalidInput = InvalidInput -> IO ()
forall a. ToCBOR a => a -> IO ()
sendPlainCbor
        , $sel:decodeInput:WsCodec :: ByteString -> Either String (ClientInput tx)
decodeInput = ApiEncoding -> ByteString -> Either String (ClientInput tx)
forall a.
(FromJSON a, FromCBOR a) =>
ApiEncoding -> ByteString -> Either String a
decodeWire ApiEncoding
CborEncoding
        , $sel:describeInput:WsCodec :: ByteString -> Text
describeInput = ApiEncoding -> ByteString -> Text
describeWire ApiEncoding
CborEncoding
        }
 where
  sendPlainJson :: ToJSON a => a -> IO ()
  sendPlainJson :: forall a. ToJSON a => a -> IO ()
sendPlainJson = Connection -> ByteString -> IO ()
forall a. WebSocketsData a => Connection -> a -> IO ()
sendTextData Connection
con (ByteString -> IO ()) -> (a -> ByteString) -> a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> ByteString
forall a. ToJSON a => a -> ByteString
Aeson.encode

  sendPlainCbor :: ToCBOR a => a -> IO ()
  sendPlainCbor :: forall a. ToCBOR a => a -> IO ()
sendPlainCbor = Connection -> ByteString -> IO ()
forall a. WebSocketsData a => Connection -> a -> IO ()
sendBinaryData Connection
con (ByteString -> IO ()) -> (a -> ByteString) -> a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize'

wsApp ::
  forall tx.
  IsChainState tx =>
  Environment ->
  Party ->
  Tracer IO APIServerLog ->
  Chain tx IO ->
  ConduitT () (TimedServerOutput tx) (ResourceT IO) () ->
  (ClientInput tx -> IO ()) ->
  -- | Read model to enhance 'Greetings' messages with 'HeadStatus'.
  Projection STM.STM (StateChanged tx) (NodeState tx) ->
  -- | Read model to enhance 'Greetings' messages with 'NetworkInfo'.
  Projection STM.STM (StateChanged tx) NetworkInfo ->
  TChan (Either (TimedServerOutput tx) (ClientMessage tx)) ->
  ServerOutputFilter tx ->
  PendingConnection ->
  IO ()
wsApp :: forall tx.
IsChainState tx =>
Environment
-> Party
-> Tracer IO APIServerLog
-> Chain tx IO
-> ConduitT () (TimedServerOutput tx) (ResourceT IO) ()
-> (ClientInput tx -> IO ())
-> Projection STM (StateChanged tx) (NodeState tx)
-> Projection STM (StateChanged tx) NetworkInfo
-> TChan (Either (TimedServerOutput tx) (ClientMessage tx))
-> ServerOutputFilter tx
-> PendingConnection
-> IO ()
wsApp Environment
env Party
party Tracer IO APIServerLog
tracer Chain tx IO
chain ConduitT () (TimedServerOutput tx) (ResourceT IO) ()
history ClientInput tx -> IO ()
callback Projection STM (StateChanged tx) (NodeState tx)
nodeStateP Projection STM (StateChanged tx) NetworkInfo
networkInfoP TChan (Either (TimedServerOutput tx) (ClientMessage tx))
responseChannel ServerOutputFilter{TimedServerOutput tx -> Text -> Bool
txContainsAddr :: TimedServerOutput tx -> Text -> Bool
$sel:txContainsAddr:ServerOutputFilter :: forall tx.
ServerOutputFilter tx -> TimedServerOutput tx -> Text -> Bool
txContainsAddr} PendingConnection
pending = do
  Tracer IO APIServerLog -> APIServerLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO APIServerLog
tracer APIServerLog
NewAPIConnection
  let queryParams :: Query
queryParams = ByteString -> Query
queryParamsOf (ByteString -> Query)
-> (RequestHead -> ByteString) -> RequestHead -> Query
forall b c a. (b -> c) -> (a -> b) -> a -> c
. RequestHead -> ByteString
requestPath (RequestHead -> Query) -> RequestHead -> Query
forall a b. (a -> b) -> a -> b
$ PendingConnection -> RequestHead
pendingRequest PendingConnection
pending
  Connection
con <- PendingConnection -> IO Connection
acceptRequest PendingConnection
pending
  TChan (Either (TimedServerOutput tx) (ClientMessage tx))
chan <- STM (TChan (Either (TimedServerOutput tx) (ClientMessage tx)))
-> IO (TChan (Either (TimedServerOutput tx) (ClientMessage tx)))
forall a. STM a -> IO a
STM.atomically (STM (TChan (Either (TimedServerOutput tx) (ClientMessage tx)))
 -> IO (TChan (Either (TimedServerOutput tx) (ClientMessage tx))))
-> STM (TChan (Either (TimedServerOutput tx) (ClientMessage tx)))
-> IO (TChan (Either (TimedServerOutput tx) (ClientMessage tx)))
forall a b. (a -> b) -> a -> b
$ TChan (Either (TimedServerOutput tx) (ClientMessage tx))
-> STM (TChan (Either (TimedServerOutput tx) (ClientMessage tx)))
forall a. TChan a -> STM (TChan a)
dupTChan TChan (Either (TimedServerOutput tx) (ClientMessage tx))
responseChannel

  let outConfig :: ServerOutputConfig
outConfig = Query -> ServerOutputConfig
mkServerOutputConfig Query
queryParams
      codec :: WsCodec tx
codec = ServerOutputConfig -> Connection -> WsCodec tx
forall tx.
IsChainState tx =>
ServerOutputConfig -> Connection -> WsCodec tx
mkWsCodec ServerOutputConfig
outConfig Connection
con

  -- api client can decide if they want to see the past history of server outputs
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Query -> Bool
shouldServeHistory Query
queryParams) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    WsCodec tx -> ServerOutputConfig -> IO ()
forwardHistory WsCodec tx
codec ServerOutputConfig
outConfig

  WsCodec tx -> ServerOutputConfig -> IO ()
forwardGreetingOnly WsCodec tx
codec ServerOutputConfig
outConfig

  Connection -> Int -> IO () -> IO () -> IO ()
forall a. Connection -> Int -> IO () -> IO a -> IO a
withPingThread Connection
con Int
30 (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    (String, IO Any) -> (String, IO Any) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (String, m b) -> m ()
raceLabelled_
      (String
"ws-con-receive-inputs", WsCodec tx -> Connection -> IO Any
receiveInputs WsCodec tx
codec Connection
con)
      (String
"ws-con-send-outputs", WsCodec tx
-> TChan (Either (TimedServerOutput tx) (ClientMessage tx))
-> ServerOutputConfig
-> IO Any
sendOutputs WsCodec tx
codec TChan (Either (TimedServerOutput tx) (ClientMessage tx))
chan ServerOutputConfig
outConfig)
 where
  -- NOTE: We will add a 'Greetings' message on each API server start. This is
  -- important to make sure the latest configured 'party' is reaching the
  -- client.
  forwardGreetingOnly :: WsCodec tx -> ServerOutputConfig -> IO ()
forwardGreetingOnly WsCodec tx
codec ServerOutputConfig
config = do
    NodeState tx
nodeState <- STM IO (NodeState tx) -> IO (NodeState tx)
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM (NodeState tx)
STM IO (NodeState tx)
getLatestNodeState
    let headState :: HeadState tx
headState = NodeState tx
nodeState.headState
    NetworkInfo
networkInfo <- STM IO NetworkInfo -> IO NetworkInfo
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM NetworkInfo
STM IO NetworkInfo
getLatestNetworkInfo
    let greetings :: Greetings tx
greetings =
          Greetings
            { $sel:me:Greetings :: Party
me = Party
party
            , $sel:headStatus:Greetings :: HeadStatus
headStatus = HeadState tx -> HeadStatus
getHeadStatus HeadState tx
headState
            , $sel:hydraHeadId:Greetings :: Maybe HeadId
hydraHeadId = HeadState tx -> Maybe HeadId
getHeadId HeadState tx
headState
            , $sel:snapshotUtxo:Greetings :: Maybe (UTxOType tx)
snapshotUtxo =
                case ServerOutputConfig
config.utxoInSnapshot of
                  WithUTxO
WithUTxO -> HeadState tx -> Maybe (UTxOType tx)
forall tx. IsTx tx => HeadState tx -> Maybe (UTxOType tx)
getSnapshotUtxo HeadState tx
headState
                  WithUTxO
WithoutUTxO -> Maybe (UTxOType tx)
forall a. Maybe a
Nothing
            , $sel:hydraNodeVersion:Greetings :: String
hydraNodeVersion = Version -> String
showVersion Version
NetworkVersions.hydraNodeVersion
            , Environment
env :: Environment
$sel:env:Greetings :: Environment
env
            , NetworkInfo
networkInfo :: NetworkInfo
$sel:networkInfo:Greetings :: NetworkInfo
networkInfo
            , $sel:chainSyncedStatus:Greetings :: SyncedStatus
chainSyncedStatus = NodeState tx -> SyncedStatus
forall tx. NodeState tx -> SyncedStatus
syncedStatus NodeState tx
nodeState
            , $sel:currentSlot:Greetings :: ChainSlot
currentSlot = NodeState tx
nodeState.chainPointTime.currentSlot
            }
    WsCodec tx
codec.sendGreetings Greetings tx
greetings

  Projection{$sel:getLatest:Projection :: forall (stm :: * -> *) event model.
Projection stm event model -> stm model
getLatest = STM (NodeState tx)
getLatestNodeState} = Projection STM (StateChanged tx) (NodeState tx)
nodeStateP
  Projection{$sel:getLatest:Projection :: forall (stm :: * -> *) event model.
Projection stm event model -> stm model
getLatest = STM NetworkInfo
getLatestNetworkInfo} = Projection STM (StateChanged tx) NetworkInfo
networkInfoP

  sendOutputs :: WsCodec tx
-> TChan (Either (TimedServerOutput tx) (ClientMessage tx))
-> ServerOutputConfig
-> IO Any
sendOutputs WsCodec tx
codec TChan (Either (TimedServerOutput tx) (ClientMessage tx))
chan ServerOutputConfig{WithAddressedTx
addressInTx :: WithAddressedTx
$sel:addressInTx:ServerOutputConfig :: ServerOutputConfig -> WithAddressedTx
addressInTx} = IO () -> IO Any
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO Any) -> IO () -> IO Any
forall a b. (a -> b) -> a -> b
$ do
    Either (TimedServerOutput tx) (ClientMessage tx)
response <- STM (Either (TimedServerOutput tx) (ClientMessage tx))
-> IO (Either (TimedServerOutput tx) (ClientMessage tx))
forall a. STM a -> IO a
STM.atomically (STM (Either (TimedServerOutput tx) (ClientMessage tx))
 -> IO (Either (TimedServerOutput tx) (ClientMessage tx)))
-> STM (Either (TimedServerOutput tx) (ClientMessage tx))
-> IO (Either (TimedServerOutput tx) (ClientMessage tx))
forall a b. (a -> b) -> a -> b
$ TChan (Either (TimedServerOutput tx) (ClientMessage tx))
-> STM (Either (TimedServerOutput tx) (ClientMessage tx))
forall a. TChan a -> STM a
readTChan TChan (Either (TimedServerOutput tx) (ClientMessage tx))
chan
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (WithAddressedTx
-> Either (TimedServerOutput tx) (ClientMessage tx) -> Bool
isAddressInTx WithAddressedTx
addressInTx Either (TimedServerOutput tx) (ClientMessage tx)
response) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
      Either (TimedServerOutput tx) (ClientMessage tx) -> IO ()
sendResponse Either (TimedServerOutput tx) (ClientMessage tx)
response
   where
    sendResponse :: Either (TimedServerOutput tx) (ClientMessage tx) -> IO ()
sendResponse = \case
      Left TimedServerOutput tx
response -> do
        WsCodec tx
codec.sendOutput TimedServerOutput tx
response
        Tracer IO APIServerLog -> APIServerLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO APIServerLog
tracer (Value -> APIServerLog
APIOutputSent (Value -> APIServerLog) -> Value -> APIServerLog
forall a b. (a -> b) -> a -> b
$ TimedServerOutput tx -> Value
forall a. ToJSON a => a -> Value
toJSON TimedServerOutput tx
response)
      Right ClientMessage tx
response -> do
        WsCodec tx
codec.sendClientMessage ClientMessage tx
response
        Tracer IO APIServerLog -> APIServerLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO APIServerLog
tracer (Value -> APIServerLog
APIOutputSent (Value -> APIServerLog) -> Value -> APIServerLog
forall a b. (a -> b) -> a -> b
$ ClientMessage tx -> Value
forall a. ToJSON a => a -> Value
toJSON ClientMessage tx
response)

  Chain{ConfirmedSnapshot tx -> Either Value ()
checkNonADAAssets :: ConfirmedSnapshot tx -> Either Value ()
$sel:checkNonADAAssets:Chain :: forall tx (m :: * -> *).
Chain tx m -> ConfirmedSnapshot tx -> Either Value ()
checkNonADAAssets} = Chain tx IO
chain

  receiveInputs :: WsCodec tx -> Connection -> IO Any
receiveInputs WsCodec tx
codec Connection
con = IO () -> IO Any
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO Any) -> IO () -> IO Any
forall a b. (a -> b) -> a -> b
$ do
    ByteString
msg <- Connection -> IO ByteString
forall a. WebSocketsData a => Connection -> IO a
receiveData Connection
con
    let receivedText :: Text
receivedText = WsCodec tx
codec.describeInput ByteString
msg
    case WsCodec tx
codec.decodeInput ByteString
msg of
      Right ClientInput tx
input -> do
        Tracer IO APIServerLog -> APIServerLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO APIServerLog
tracer (Value -> APIServerLog
APIInputReceived (Value -> APIServerLog) -> Value -> APIServerLog
forall a b. (a -> b) -> a -> b
$ ClientInput tx -> Value
forall a. ToJSON a => a -> Value
toJSON ClientInput tx
input)
        case ClientInput tx
input of
          ClientInput tx
SafeClose -> do
            NodeState tx
nodeState <- STM IO (NodeState tx) -> IO (NodeState tx)
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM (NodeState tx)
STM IO (NodeState tx)
getLatestNodeState
            case HeadState tx -> Maybe (ConfirmedSnapshot tx)
forall tx. HeadState tx -> Maybe (ConfirmedSnapshot tx)
HeadState.getOpenStateConfirmedSnapshot NodeState tx
nodeState.headState of
              Maybe (ConfirmedSnapshot tx)
Nothing -> ClientInput tx -> IO ()
callback ClientInput tx
input
              Just ConfirmedSnapshot tx
confirmedSnapshot ->
                case ConfirmedSnapshot tx -> Either Value ()
checkNonADAAssets ConfirmedSnapshot tx
confirmedSnapshot of
                  Left Value
nonADAValue -> do
                    let errorStr :: String
errorStr = String
"Cannot SafeClose with non-ADA assets present: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Value -> String
forall b a. (Show a, IsString b) => a -> b
show Value
nonADAValue
                    WsCodec tx
codec.sendInvalidInput (InvalidInput -> IO ()) -> InvalidInput -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Text -> InvalidInput
InvalidInput String
errorStr Text
receivedText
                    Tracer IO APIServerLog -> APIServerLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO APIServerLog
tracer (String -> Text -> APIServerLog
APIInvalidInput String
errorStr Text
receivedText)
                  Right ()
_ -> ClientInput tx -> IO ()
callback ClientInput tx
input
          ClientInput tx
_ -> ClientInput tx -> IO ()
callback ClientInput tx
input
      Left String
e -> do
        -- XXX(AB): toStrict might be problematic as it implies consuming the full
        -- message to memory
        WsCodec tx
codec.sendInvalidInput (InvalidInput -> IO ()) -> InvalidInput -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Text -> InvalidInput
InvalidInput String
e Text
receivedText
        Tracer IO APIServerLog -> APIServerLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO APIServerLog
tracer (String -> Text -> APIServerLog
APIInvalidInput String
e Text
receivedText)

  forwardHistory :: WsCodec tx -> ServerOutputConfig -> IO ()
forwardHistory WsCodec tx
codec ServerOutputConfig{WithAddressedTx
$sel:addressInTx:ServerOutputConfig :: ServerOutputConfig -> WithAddressedTx
addressInTx :: WithAddressedTx
addressInTx} =
    ConduitT () Void (ResourceT IO) () -> IO ()
forall (m :: * -> *) r.
MonadUnliftIO m =>
ConduitT () Void (ResourceT m) r -> m r
runConduitRes (ConduitT () Void (ResourceT IO) () -> IO ())
-> ConduitT () Void (ResourceT IO) () -> IO ()
forall a b. (a -> b) -> a -> b
$ ConduitT () (TimedServerOutput tx) (ResourceT IO) ()
history ConduitT () (TimedServerOutput tx) (ResourceT IO) ()
-> ConduitT (TimedServerOutput tx) Void (ResourceT IO) ()
-> ConduitT () Void (ResourceT IO) ()
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| (TimedServerOutput tx -> Bool)
-> ConduitT
     (TimedServerOutput tx) (TimedServerOutput tx) (ResourceT IO) ()
forall (m :: * -> *) a. Monad m => (a -> Bool) -> ConduitT a a m ()
filter (WithAddressedTx
-> Either (TimedServerOutput tx) (ClientMessage tx) -> Bool
isAddressInTx WithAddressedTx
addressInTx (Either (TimedServerOutput tx) (ClientMessage tx) -> Bool)
-> (TimedServerOutput tx
    -> Either (TimedServerOutput tx) (ClientMessage tx))
-> TimedServerOutput tx
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TimedServerOutput tx
-> Either (TimedServerOutput tx) (ClientMessage tx)
forall a b. a -> Either a b
Left) ConduitT
  (TimedServerOutput tx) (TimedServerOutput tx) (ResourceT IO) ()
-> ConduitT (TimedServerOutput tx) Void (ResourceT IO) ()
-> ConduitT (TimedServerOutput tx) Void (ResourceT IO) ()
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| (TimedServerOutput tx -> ResourceT IO ())
-> ConduitT (TimedServerOutput tx) Void (ResourceT IO) ()
forall (m :: * -> *) a o.
Monad m =>
(a -> m ()) -> ConduitT a o m ()
mapM_C (IO () -> ResourceT IO ()
forall a. IO a -> ResourceT IO a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> ResourceT IO ())
-> (TimedServerOutput tx -> IO ())
-> TimedServerOutput tx
-> ResourceT IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. WsCodec tx
codec.sendOutput)

  isAddressInTx :: WithAddressedTx
-> Either (TimedServerOutput tx) (ClientMessage tx) -> Bool
isAddressInTx WithAddressedTx
addressInTx = \case
    Left TimedServerOutput tx
tx -> TimedServerOutput tx -> Bool
checkAddress TimedServerOutput tx
tx
    Right ClientMessage tx
_ -> Bool
True
   where
    checkAddress :: TimedServerOutput tx -> Bool
checkAddress TimedServerOutput tx
tx =
      case WithAddressedTx
addressInTx of
        WithAddressedTx Text
addr -> TimedServerOutput tx -> Text -> Bool
txContainsAddr TimedServerOutput tx
tx Text
addr
        WithAddressedTx
WithoutAddressedTx -> Bool
True

  -- \| Get the content of 'headStatus' field in 'Greetings' message from the full 'HeadState'.
  getHeadStatus :: HeadState tx -> HeadStatus
  getHeadStatus :: HeadState tx -> HeadStatus
getHeadStatus = \case
    HeadState.Idle{} -> HeadStatus
Idle
    HeadState.Open{} -> HeadStatus
Open
    HeadState.Closed ClosedState{Bool
$sel:readyToFanoutSent:ClosedState :: forall tx. ClosedState tx -> Bool
readyToFanoutSent :: Bool
readyToFanoutSent}
      | Bool
readyToFanoutSent -> HeadStatus
FanoutPossible
      | Bool
otherwise -> HeadStatus
Closed
    HeadState.FanoutProgress{} -> HeadStatus
FanningOut

  getHeadId :: HeadState tx -> Maybe HeadId
  getHeadId :: HeadState tx -> Maybe HeadId
getHeadId = \case
    HeadState.Idle{} -> Maybe HeadId
forall a. Maybe a
Nothing
    HeadState.Open OpenState{HeadId
headId :: HeadId
$sel:headId:OpenState :: forall tx. OpenState tx -> HeadId
headId} -> HeadId -> Maybe HeadId
forall a. a -> Maybe a
Just HeadId
headId
    HeadState.Closed ClosedState{HeadId
headId :: HeadId
$sel:headId:ClosedState :: forall tx. ClosedState tx -> HeadId
headId} -> HeadId -> Maybe HeadId
forall a. a -> Maybe a
Just HeadId
headId
    HeadState.FanoutProgress PartialFanoutState{HeadId
headId :: HeadId
$sel:headId:PartialFanoutState :: forall tx. PartialFanoutState tx -> HeadId
headId} -> HeadId -> Maybe HeadId
forall a. a -> Maybe a
Just HeadId
headId

-- | The query parameters of a websocket connection request path.
--
-- NOTE: a malformed query string yields no parameters here, and hence the
-- default output config. modern-uri (used here previously) raised a parse
-- exception during connection setup instead.
queryParamsOf :: ByteString -> Query
queryParamsOf :: ByteString -> Query
queryParamsOf = ByteString -> Query
parseQuery (ByteString -> Query)
-> (ByteString -> ByteString) -> ByteString -> Query
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Char -> Bool) -> ByteString -> ByteString
BS8.dropWhile (Char -> Char -> Bool
forall a. Eq a => a -> a -> Bool
/= Char
'?')

-- | Decide what a client wants to see, from the query string of its connection.
mkServerOutputConfig :: Query -> ServerOutputConfig
mkServerOutputConfig :: Query -> ServerOutputConfig
mkServerOutputConfig Query
qp =
  ServerOutputConfig
    { $sel:utxoInSnapshot:ServerOutputConfig :: WithUTxO
utxoInSnapshot = Query -> WithUTxO
decideOnUTxODisplay Query
qp
    , $sel:addressInTx:ServerOutputConfig :: WithAddressedTx
addressInTx = Query -> WithAddressedTx
decideOnAddressDisplay Query
qp
    , $sel:encoding:ServerOutputConfig :: ApiEncoding
encoding = Query -> ApiEncoding
decideOnEncoding Query
qp
    }

decideOnEncoding :: Query -> ApiEncoding
decideOnEncoding :: Query -> ApiEncoding
decideOnEncoding Query
qp =
  if (ByteString
"encoding", ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"cbor") (ByteString, Maybe ByteString) -> Query -> Bool
forall (f :: * -> *) a.
(Foldable f, DisallowElem f, Eq a) =>
a -> f a -> Bool
`elem` Query
qp then ApiEncoding
CborEncoding else ApiEncoding
JsonEncoding

decideOnUTxODisplay :: Query -> WithUTxO
decideOnUTxODisplay :: Query -> WithUTxO
decideOnUTxODisplay Query
qp =
  if (ByteString
"snapshot-utxo", ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"no") (ByteString, Maybe ByteString) -> Query -> Bool
forall (f :: * -> *) a.
(Foldable f, DisallowElem f, Eq a) =>
a -> f a -> Bool
`elem` Query
qp then WithUTxO
WithoutUTxO else WithUTxO
WithUTxO

decideOnAddressDisplay :: Query -> WithAddressedTx
decideOnAddressDisplay :: Query -> WithAddressedTx
decideOnAddressDisplay Query
qp =
  -- NOTE: takes the first 'address' that actually carries a value. A valueless
  -- '?address' is skipped rather than disabling the filter: modern-uri (used
  -- here previously) parsed that as a QueryFlag, which the old lookup ignored,
  -- so '?address&address=addr1...' still filtered. An empty '?address=' is
  -- skipped for the same reason: the filter compares addresses exactly, so
  -- keeping it would match nothing and the client would silently see no
  -- transaction outputs at all.
  case [ByteString] -> Maybe ByteString
forall a. [a] -> Maybe a
listToMaybe [ByteString
v | (ByteString
"address", Just ByteString
v) <- Query
qp, Bool -> Bool
not (ByteString -> Bool
BS8.null ByteString
v)] of
    Just ByteString
v -> Text -> WithAddressedTx
WithAddressedTx (ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 ByteString
v)
    Maybe ByteString
Nothing -> WithAddressedTx
WithoutAddressedTx

shouldServeHistory :: Query -> Bool
shouldServeHistory :: Query -> Bool
shouldServeHistory Query
qp =
  (ByteString
"history", ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
"yes") (ByteString, Maybe ByteString) -> Query -> Bool
forall (f :: * -> *) a.
(Foldable f, DisallowElem f, Eq a) =>
a -> f a -> Bool
`elem` Query
qp