{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeFamilies #-}
{-# LANGUAGE UndecidableInstances #-}
{-# LANGUAGE NoImplicitPrelude #-}
module Data.Morpheus.Subscriptions.Stream
( toOutStream,
runStreamWS,
runStreamHTTP,
ApiContext (..),
Input (..),
Output,
API (..),
PUB,
SUB,
)
where
import Control.Monad.Except (throwError)
import Data.ByteString.Lazy.Char8 (ByteString)
import Data.Morpheus.App.Internal.Resolving
( Channel,
ResponseEvent (..),
ResponseStream,
Result (..),
ResultT (..),
runResultT,
)
import Data.Morpheus.Subscriptions.Apollo
( ApolloAction (..),
ApolloMessageType (..),
apolloFormat,
toApolloResponse,
)
import Data.Morpheus.Subscriptions.ClientConnectionStore
( ClientConnectionStore,
SessionID (..),
Updates (..),
endSession,
insertConnection,
startSession,
)
import Data.Morpheus.Subscriptions.Event (Event (..))
import Data.Morpheus.Types.IO
( GQLRequest (..),
GQLResponse (..),
)
import Data.Morpheus.Types.Internal.AST
( GQLError,
VALID,
Value (..),
)
import Data.UUID (UUID)
import Relude hiding (ByteString)
data API = PUB | SUB
type SUB = 'SUB
type PUB = 'PUB
data
Input
(api :: API)
where
InitConnection :: UUID -> Input SUB
Request :: GQLRequest -> Input PUB
run :: ApiContext SUB e m -> Updates e m -> m ()
run SubContext {updateStore} (Updates changes) = updateStore changes
data ApiContext (api :: API) event (m :: Type -> Type) where
PubContext ::
{ eventPublisher :: event -> m ()
} ->
ApiContext PUB event m
SubContext ::
{ listener :: m ByteString,
callback :: ByteString -> m (),
updateStore :: (ClientConnectionStore event m -> ClientConnectionStore event m) -> m ()
} ->
ApiContext SUB event m
data WSOutputEvent e m
= WSUpdate (Updates e m)
| WSMessage ByteString
data
Output
(api :: API)
(e :: Type)
(m :: Type -> Type)
where
SubOutput ::
{ streamWS :: ApiContext SUB e m -> m (Either ByteString [WSOutputEvent e m])
} ->
Output SUB e m
PubOutput ::
{ streamHTTP :: ApiContext PUB e m -> m GQLResponse
} ->
Output PUB e m
handleResponseStream ::
( Monad m,
Eq (Channel e),
Hashable (Channel e)
) =>
SessionID ->
ResponseStream e m (Value VALID) ->
Output SUB e m
handleResponseStream session (ResultT res) =
SubOutput $ const $ unfoldR <$> res
where
execute Publish {} =
apolloError
["websocket can only handle subscriptions, not mutations"]
execute (Subscribe ch subRes) =
Right . WSUpdate $ startSession ch subRes session
--------------------------
unfoldR Success {result = (events, _)} =
traverse execute events
unfoldR Failure {errors} =
apolloError (toList errors)
--------------------------
apolloError :: [GQLError] -> Either ByteString a
apolloError = Left . toApolloResponse GqlError (Just $ sid session) . Just . Errors
handleWSRequest ::
( Monad m,
Functor m,
Eq ch,
Hashable ch
) =>
( GQLRequest ->
ResponseStream (Event ch con) m (Value VALID)
) ->
UUID ->
ByteString ->
Output SUB (Event ch con) m
handleWSRequest gqlApp clientId = handle . apolloFormat
where
-- handle :: Applicative m => Validation ApolloAction -> Stream SUB e m
handle = either (liftWS . Left) handleAction
--------------------------------------------------
-- handleAction :: ApolloAction -> Stream SUB e m
handleAction ConnectionInit = do
liftWS $ Right [WSMessage $ toApolloResponse GqlConnectionAck Nothing Nothing]
handleAction Ping = do
liftWS $ Right [WSMessage $ toApolloResponse GqlPong Nothing Nothing]
handleAction (SessionStart sessionId request) =
handleResponseStream (SessionID clientId sessionId) (gqlApp request)
handleAction (SessionStop sessionId) =
liftWS
$ Right [WSUpdate $ endSession (SessionID clientId sessionId)]
liftWS ::
(Applicative m) =>
Either ByteString [WSOutputEvent e m] ->
Output SUB e m
liftWS = SubOutput . const . pure
runStreamWS ::
(Monad m) =>
ApiContext SUB e m ->
Output SUB e m ->
m ()
runStreamWS scope@SubContext {callback} SubOutput {streamWS} =
streamWS scope
>>= either callback (traverse_ eventRunner)
where
-- eventRunner :: Monad m => WSOutputEvent e m -> m ()
eventRunner (WSUpdate updates) = run scope updates
eventRunner (WSMessage msg) = callback msg
runStreamHTTP ::
(Monad m) =>
ApiContext PUB e m ->
Output PUB e m ->
m GQLResponse
runStreamHTTP scope PubOutput {streamHTTP} =
streamHTTP scope
toOutStream ::
( Monad m,
Eq ch,
Hashable ch
) =>
( GQLRequest ->
ResponseStream (Event ch con) m (Value VALID)
) ->
Input api ->
Output api (Event ch con) m
toOutStream app (InitConnection clientId) =
SubOutput handle
where
handle ws@SubContext {listener, callback} = do
let runS (SubOutput x) = x ws
bla <- listener >>= runS . handleWSRequest app clientId
pure $ ((WSUpdate $ Updates (insertConnection clientId callback)) :) <$> bla
toOutStream app (Request req) =
PubOutput $ handleResponseHTTP (app req)
handleResponseHTTP ::
( Monad m
) =>
ResponseStream e m (Value VALID) ->
ApiContext PUB e m ->
m GQLResponse
handleResponseHTTP
res
PubContext {eventPublisher} = runResultT (handleRes res execute) >>= runResult
where
runResult Success {result = (events, result)} = traverse_ eventPublisher events $> Data result
runResult Failure {errors} = pure $ Errors $ toList errors
execute (Publish event) = pure event
execute Subscribe {} = throwError "http server can't handle subscription"
handleRes ::
(Monad m) =>
ResponseStream e m a ->
(ResponseEvent e m -> ResultT e' m e') ->
ResultT e' m a
handleRes res execute = ResultT $ runResultT res >>= runResultT . unfoldRes execute
unfoldRes ::
(Monad m) =>
(e -> ResultT e' m e') ->
Result GQLError ([e], a) ->
ResultT e' m a
unfoldRes execute Success {result = (events, result), ..} = traverse execute events >>= packResultT
where
packResultT events' = ResultT $ pure $ Success {result = (events', result), ..}
unfoldRes _ Failure {errors} = ResultT $ pure $ Failure {errors}