fix remote queries/mutations to work over websocket (fix hasura#1619) (hasura#1621)

Author: ecthiender

server/src-lib/Hasura/GraphQL/Transport/WebSocket.hs

@@ -26,6 +26,7 @@ import           Control.Concurrent                          (threadDelay)
 import           Data.ByteString                             (ByteString)
 import qualified Data.IORef                                  as IORef
 
+import           Hasura.GraphQL.Context                      (GCtx)
 import           Hasura.GraphQL.Resolve                      (resolveSelSet)
 import           Hasura.GraphQL.Resolve.Context              (LazyRespTx)
 import qualified Hasura.GraphQL.Resolve.LiveQuery            as LQ
@@ -239,20 +240,13 @@ onStart serverEnv wsConn (StartMsg opId q) msgRaw = catchAndIgnore $ do
   either (\(QErr _ _ err _ _) -> withComplete $ sendConnErr err) return res'
 
   case typeLocs of
-    [] -> runHasuraQ userInfo gCtx queryParts
-
+    []          -> runHasuraQ userInfo gCtx queryParts
     (typeLoc:_) -> case typeLoc of
-      VT.HasuraType ->
-        runHasuraQ userInfo gCtx queryParts
-      VT.RemoteType _ rsi -> do
-        when (G._todType opDef == G.OperationTypeSubscription) $
-          withComplete $ sendConnErr "subscription to remote server is not supported"
-        resp <- runExceptT $ TH.runRemoteGQ httpMgr userInfo reqHdrs
-                             msgRaw rsi opDef
-        either postExecErr sendSuccResp resp
-        sendCompleted
+      VT.HasuraType       -> runHasuraQ userInfo gCtx queryParts
+      VT.RemoteType _ rsi -> runRemoteQ userInfo reqHdrs opDef rsi
 
   where
+    runHasuraQ :: UserInfo -> GCtx -> QueryParts -> ExceptT () IO ()
     runHasuraQ userInfo gCtx queryParts = do
       (opTy, fields) <- either (withComplete . preExecErr) return $
                         runReaderT (validateGQ queryParts) gCtx
@@ -270,8 +264,30 @@ onStart serverEnv wsConn (StartMsg opId q) msgRaw = catchAndIgnore $ do
           either postExecErr sendSuccResp resp
           sendCompleted
 
+    runRemoteQ :: UserInfo -> [H.Header]
+               -> G.TypedOperationDefinition -> RemoteSchemaInfo
+               -> ExceptT () IO ()
+    runRemoteQ userInfo reqHdrs opDef rsi = do
+      when (G._todType opDef == G.OperationTypeSubscription) $
+        withComplete $ preExecErr $
+        err400 NotSupported "subscription to remote server is not supported"
+
+      -- if it's not a subscription, use HTTP to execute the query on the remote
+      -- server
+      -- try to parse the (apollo protocol) websocket frame and get only the
+      -- payload
+      sockPayload <- onLeft (J.eitherDecode msgRaw) $
+        const $ withComplete $ preExecErr $
+        err500 Unexpected "invalid websocket payload"
+      let payload = J.encode $ _wpPayload sockPayload
+      resp <- runExceptT $ TH.runRemoteGQ httpMgr userInfo reqHdrs
+              payload rsi opDef
+      either postExecErr sendSuccResp resp
+      sendCompleted
+
 
     WSServerEnv logger _ runTx lqMap gCtxMapRef httpMgr _ sqlGenCtx = serverEnv
+
     wsId = WS.getWSId wsConn
     WSConnData userInfoR opMap = WS.getData wsConn
 
@@ -429,3 +445,18 @@ createWSServerApp authMode serverEnv =
       (onConn (_wseLogger serverEnv) (_wseCorsPolicy serverEnv))
       (onMessage authMode serverEnv)
       (onClose (_wseLogger serverEnv) $ _wseLiveQMap serverEnv)
+
+
+-- | TODO:
+-- | The following ADT is required so that we can parse the incoming websocket
+-- | frame, and only pick the payload, for remote schema queries.
+-- | Ideally we should use `StartMsg` from Websocket.Protocol, but as
+-- | `GraphQLRequest` doesn't have a ToJSON instance we are using our own type to
+-- | get only the payload
+data WebsocketPayload
+  = WebsocketPayload
+  { _wpId      :: !Text
+  , _wpType    :: !Text
+  , _wpPayload :: !J.Value
+  } deriving (Show, Eq)
+$(J.deriveJSON (J.aesonDrop 3 J.snakeCase) ''WebsocketPayload)

server/tests-py/test_schema_stitching.py

@@ -1,9 +1,14 @@
 #!/usr/bin/env python3
 
-import pytest
+import string
+import random
 import yaml
+import json
+import queue
 import requests
 
+import pytest
+
 from validate import check_query_f, check_query
 
 
@@ -90,7 +95,8 @@ def test_add_remote_schema_with_interfaces(self, hge_ctx):
         st_code, resp = hge_ctx.v1q(q)
         assert st_code == 200, resp
         check_query_f(hge_ctx, self.dir + '/character_interface_query.yaml')
-        hge_ctx.v1q({"type": "remove_remote_schema", "args": {"name": "my remote interface one"}})
+        hge_ctx.v1q({"type": "remove_remote_schema",
+                     "args": {"name": "my remote interface one"}})
         assert st_code == 200, resp
 
     def test_add_remote_schema_with_interface_err_empty_fields_list(self, hge_ctx):
@@ -102,23 +108,30 @@ def test_add_remote_schema_err_unknown_interface(self, hge_ctx):
         check_query_f(hge_ctx, self.dir + '/add_remote_schema_err_unknown_interface.yaml')
 
     def test_add_remote_schema_with_interface_err_missing_field(self, hge_ctx):
-        """add a remote schema where an object implementing an interface does not have a field defined in the interface"""
+        """ add a remote schema where an object implementing an interface does
+        not have a field defined in the interface """
         check_query_f(hge_ctx, self.dir + '/add_remote_schema_err_missing_field.yaml')
 
     def test_add_remote_schema_with_interface_err_wrong_field_type(self, hge_ctx):
-        """add a remote schema where an object implementing an interface have a field with the same name as in the interface, but of different type"""
+        """add a remote schema where an object implementing an interface have a
+        field with the same name as in the interface, but of different type"""
         check_query_f(hge_ctx, self.dir + '/add_remote_schema_with_iface_err_wrong_field_type.yaml')
 
     def test_add_remote_schema_with_interface_err_missing_arg(self, hge_ctx):
-        """add a remote schema where a field of an object implementing an interface does not have the argument defined in the same field of interface"""
+        """add a remote schema where a field of an object implementing an
+        interface does not have the argument defined in the same field of
+        interface"""
         check_query_f(hge_ctx, self.dir + '/add_remote_schema_err_missing_arg.yaml')
 
     def test_add_remote_schema_with_interface_err_wrong_arg_type(self, hge_ctx):
-        """add a remote schema where the argument of a field of an object implementing the interface does not have the same type as the argument defined in the field of interface"""
+        """add a remote schema where the argument of a field of an object
+        implementing the interface does not have the same type as the argument
+        defined in the field of interface"""
         check_query_f(hge_ctx, self.dir + '/add_remote_schema_iface_err_wrong_arg_type.yaml')
 
     def test_add_remote_schema_with_interface_err_extra_non_null_arg(self, hge_ctx):
-        """add a remote schema with a field of an object implementing interface having extra non_null argument"""
+        """add a remote schema with a field of an object implementing interface
+        having extra non_null argument"""
         check_query_f(hge_ctx, self.dir + '/add_remote_schema_with_iface_err_extra_non_null_arg.yaml')
 
     def test_add_remote_schema_with_union(self, hge_ctx):
@@ -226,6 +239,88 @@ def test_add_schema_same_type_containing_same_scalar(self, hge_ctx):
         assert st_code == 200, resp
 
 
+class TestRemoteSchemaQueriesOverWebsocket:
+    dir = 'queries/remote_schemas'
+    teardown = {"type": "clear_metadata", "args": {}}
+
+    @pytest.fixture(autouse=True)
+    def transact(self, hge_ctx):
+        st_code, resp = hge_ctx.v1q_f('queries/remote_schemas/tbls_setup.yaml')
+        assert st_code == 200, resp
+        yield
+        # teardown
+        st_code, resp = hge_ctx.v1q_f('queries/remote_schemas/tbls_teardown.yaml')
+        assert st_code == 200, resp
+        st_code, resp = hge_ctx.v1q(self.teardown)
+        assert st_code == 200, resp
+
+    def _init(self, hge_ctx):
+        payload = {'type': 'connection_init', 'payload': {}}
+        if hge_ctx.hge_key:
+            payload['payload']['headers'] = {
+                'x-hasura-admin-secret': hge_ctx.hge_key
+            }
+        hge_ctx.ws.send(json.dumps(payload))
+        ev = hge_ctx.get_ws_event(3)
+        assert ev['type'] == 'connection_ack', ev
+
+    def _stop(self, hge_ctx, _id):
+        data = {'id': _id, 'type': 'stop'}
+        hge_ctx.ws.send(json.dumps(data))
+        ev = hge_ctx.get_ws_event(3)
+        assert ev['type'] == 'complete', ev
+
+    def _send_query(self, hge_ctx, query):
+        self._init(hge_ctx)
+        _id = gen_id()
+        frame = {
+            'id': _id,
+            'type': 'start',
+            'payload': {'query': query},
+        }
+        if hge_ctx.hge_key:
+            frame['payload']['headers'] = {
+                'x-hasura-admin-secret': hge_ctx.hge_key
+            }
+        hge_ctx.ws.send(json.dumps(frame))
+        return _id
+
+    def test_remote_query(self, hge_ctx):
+        self._init(hge_ctx)
+        query = """
+        query {
+          user(id: 2) {
+            id
+            username
+          }
+        }
+        """
+        _id = self._send_query(hge_ctx, query)
+        ev = hge_ctx.get_ws_event(3)
+        assert ev['type'] == 'data' and ev['id'] == _id, ev
+        assert ev['payload']['data']['data']['user']['username'] == 'john'
+        self._stop(hge_ctx, _id)
+
+    def test_remote_mutation(self, hge_ctx):
+        self._init(hge_ctx)
+        query = """
+        mutation {
+          createUser(id: 42, username: "foobar") {
+            user {
+              id
+              username
+            }
+          }
+        }
+        """
+        _id = self._send_query(hge_ctx, query)
+        ev = hge_ctx.get_ws_event(3)
+        assert ev['type'] == 'data' and ev['id'] == _id, ev
+        assert ev['payload']['data']['data']['createUser']['user']['id'] == 42
+        assert ev['payload']['data']['data']['createUser']['user']['username'] == 'foobar'
+        self._stop(hge_ctx, _id)
+
+
 class TestAddRemoteSchemaCompareRootQueryFields:
 
     remote = 'http://localhost:5000/default-value-echo-graphql'
@@ -250,35 +345,35 @@ def test_schema_check_arg_default_values_and_field_and_arg_types(self, hge_ctx):
         introspect_remote = resp.json()
         assert resp.status_code == 200, introspect_remote
         remote_root_ty_info = get_query_root_info(introspect_remote)
-        hasura_root_ty_Info = get_query_root_info(introspect_hasura)
-        hasFld=dict()
+        hasura_root_ty_info = get_query_root_info(introspect_hasura)
+        has_fld = dict()
         for fr in remote_root_ty_info['fields']:
-            hasFld[fr['name']] = False
-            for fh in filter(lambda f: f['name'] == fr['name'], hasura_root_ty_Info['fields']):
-                hasFld[fr['name']] = True
+            has_fld[fr['name']] = False
+            for fh in filter(lambda f: f['name'] == fr['name'], hasura_root_ty_info['fields']):
+                has_fld[fr['name']] = True
                 assert fr['type'] == fh['type'], yaml.dump({
                     'error' : 'Types do not match for fld ' + fr['name'],
                     'remote_type' : fr['type'],
                     'hasura_type' : fh['type']
                 })
-                hasArg=dict()
+                has_arg = dict()
                 for ar in fr['args']:
-                    arPath = fr['name'] + '(' + ar['name'] + ':)'
-                    hasArg[arPath] = False
+                    arg_path = fr['name'] + '(' + ar['name'] + ':)'
+                    has_arg[arg_path] = False
                     for ah in filter(lambda a: a['name'] == ar['name'], fh['args']):
-                        hasArg[arPath] = True
+                        has_arg[arg_path] = True
                         assert ar['type'] == ah['type'], yaml.dump({
-                            'error' : 'Types do not match for arg ' + arPath,
+                            'error' : 'Types do not match for arg ' + arg_path,
                             'remote_type' : ar['type'],
                             'hasura_type' : ah['type']
                         })
                         assert ar['defaultValue'] == ah['defaultValue'], yaml.dump({
-                            'error' : 'Default values do not match for arg ' + arPath,
+                            'error' : 'Default values do not match for arg ' + arg_path,
                             'remote_default_value' : ar['defaultValue'],
                             'hasura_default_value' : ah['defaultValue']
                         })
-                    assert hasArg[arPath], 'Argument ' + arPath + ' in the remote schema root query type not found in Hasura schema'
-            assert hasFld[fr['name']], 'Field ' + fr['name'] + ' in the remote shema root query type not found in Hasura schema'
+                    assert has_arg[arg_path], 'Argument ' + arg_path + ' in the remote schema root query type not found in Hasura schema'
+            assert has_fld[fr['name']], 'Field ' + fr['name'] + ' in the remote shema root query type not found in Hasura schema'
 
 
 #    def test_remote_query_variables(self, hge_ctx):
@@ -297,7 +392,7 @@ def _filter(f, l):
 
 def get_query_root_info(res):
     root_ty_name = res['data']['__schema']['queryType']['name']
-    return list(filter(lambda ty: ty['name'] == root_ty_name, get_types(res) ))[0]
+    return _filter(lambda ty: ty['name'] == root_ty_name, get_types(res))[0]
 
 def get_types(res):
     return res['data']['__schema']['types']
@@ -322,3 +417,6 @@ def check_introspection_result(res, types, node_names):
             satisfy_node = False
 
     return satisfy_node and satisfy_ty
+
+def gen_id(size=6, chars=string.ascii_letters + string.digits):
+    return ''.join(random.choice(chars) for _ in range(size))