Graphql server improvements cherry picked from 3js branch

Author: doombubbles

spellsource-client/src/unity

@@ -20,9 +20,28 @@ type Handler = (req: Request, res: Response, next: NextFunction) => void;
 export const setupApolloServer = async (app: Application, httpServer: Server): Promise<ApolloServer> => {
   const schema = await createFullSchema();
 
+  // Strawberry Shake passes the auth token via `?accessToken=<jwt>` on the WS URL
+  // (see SpellsourceClient.ConfigureWebSocketClient on the client). Extract it during
+  // upgrade so subscription operations can forward it to the Java backend.
+  const tokenFromUrl = (url: string | undefined): string | undefined => {
+    if (!url) return undefined;
+    const parsed = new URL(url, "http://localhost");
+    const token = parsed.searchParams.get("accessToken");
+    return token && token !== "0" ? token : undefined;
+  };
+
   // Modern graphql-transport-ws protocol (graphql-ws library)
   const graphqlWsServer = new WebSocketServer({ noServer: true });
-  const serverCleanup = useServer({ schema }, graphqlWsServer);
+  const serverCleanup = useServer(
+    {
+      schema,
+      // The context returned here is passed as `executionRequest.context` to
+      // schema executors — including the wsExecutor in spellsource.ts which
+      // reads `token` to forward auth to the Java backend's connection_init.
+      context: (ctx) => ({ token: tokenFromUrl(ctx.extra.request.url) }),
+    },
+    graphqlWsServer,
+  );
 
   // Legacy graphql-ws subprotocol (subscriptions-transport-ws library) — for Strawberry Shake clients
   const legacyWsServer = new WebSocketServer({ noServer: true });
@@ -31,6 +50,19 @@ export const setupApolloServer = async (app: Application, httpServer: Server): P
       schema,
       execute: execute as any,
       subscribe: subscribe as any,
+      onConnect: (_params: any, _ws: any, connectionContext: any) => {
+        // onConnect's return value becomes the operation context for execute/subscribe.
+        const token = tokenFromUrl(connectionContext?.request?.url);
+        console.log(`[legacy-ws] client connected, hasToken=${!!token}`);
+        return { token };
+      },
+      onOperation: (_msg: any, params: any) => {
+        console.log("[legacy-ws] operation started:", JSON.stringify(params.query?.substring(0, 200)));
+        return params;
+      },
+      onDisconnect: () => {
+        console.log("[legacy-ws] client disconnected");
+      },
     },
     legacyWsServer,
   );
@@ -44,12 +76,20 @@ export const setupApolloServer = async (app: Application, httpServer: Server): P
       ? protocolHeader
       : protocolHeader?.split(",").map((p) => p.trim());
 
+    console.log(`[ws-upgrade] protocols requested: ${JSON.stringify(protocols)}`);
+
     const wss =
       protocols?.includes(GRAPHQL_WS) && !protocols.includes(GRAPHQL_TRANSPORT_WS_PROTOCOL)
         ? legacyWsServer
         : graphqlWsServer;
 
+    console.log(`[ws-upgrade] routing to: ${wss === legacyWsServer ? "legacy" : "modern"}`);
+
     wss.handleUpgrade(req, socket, head, (ws) => {
+      // Log raw messages from the client
+      ws.on("message", (data) => {
+        console.log(`[ws-message] ${data.toString().substring(0, 500)}`);
+      });
       wss.emit("connection", ws, req);
     });
   });

spellsource-graphql/src/apollo-server.ts

@@ -1,9 +1,14 @@
 import { buildHTTPExecutor } from "@graphql-tools/executor-http";
 import { schemaFromExecutor, wrapSchema } from "@graphql-tools/wrap";
+import { Executor, observableToAsyncIterable } from "@graphql-tools/utils";
+import { getOperationAST, print } from "graphql";
+import { createClient } from "graphql-ws";
+import WebSocket from "ws";
 import { spellsourceHost, spellsourcePort } from "../config";
 import { AuthRequest } from "../auth";
 
-const executor = buildHTTPExecutor({
+// HTTP executor — used for queries and mutations.
+const httpExecutor = buildHTTPExecutor({
   endpoint: `http://${spellsourceHost}:${spellsourcePort}/graphql`,
   fetch: (url, init, context, info) => {
     const req = context as AuthRequest;
@@ -18,8 +23,65 @@ const executor = buildHTTPExecutor({
   },
 });
 
+// WebSocket executor — used for subscriptions. Forwards subscription operations
+// to the Java backend's graphql-transport-ws endpoint at /graphql, opening a fresh
+// graphql-ws connection per subscription so the auth token can be passed via
+// connection_init payload (which the backend's GraphQLWSHandler reads).
+const wsExecutor: Executor = (executionRequest) => {
+  const req = executionRequest.context as AuthRequest | undefined;
+  const token = req?.token;
+
+  const client = createClient({
+    url: `ws://${spellsourceHost}:${spellsourcePort}/graphql`,
+    webSocketImpl: WebSocket,
+    connectionParams: token ? { Authorization: `Bearer ${token}` } : {},
+    lazy: false,
+  });
+
+  return observableToAsyncIterable({
+    subscribe: (observer) => {
+      const unsubscribe = client.subscribe(
+        {
+          query: print(executionRequest.document),
+          variables: executionRequest.variables as Record<string, unknown>,
+          operationName: executionRequest.operationName,
+          extensions: executionRequest.extensions as Record<string, unknown>,
+        },
+        {
+          next: (data) => observer.next?.(data as never),
+          error: (err) => {
+            if (!observer.error) return;
+            if (err instanceof Error) {
+              observer.error(err);
+            } else if (Array.isArray(err)) {
+              observer.error(new Error(err.map((e) => (e as { message: string }).message).join(", ")));
+            } else {
+              observer.error(new Error(`Subscription error: ${JSON.stringify(err)}`));
+            }
+          },
+          complete: () => observer.complete?.(),
+        },
+      );
+      return {
+        unsubscribe: () => {
+          unsubscribe();
+          void client.dispose();
+        },
+      };
+    },
+  });
+};
+
+const executor: Executor = (executionRequest) => {
+  const operationAST = getOperationAST(executionRequest.document, executionRequest.operationName);
+  if (operationAST?.operation === "subscription") {
+    return wsExecutor(executionRequest);
+  }
+  return httpExecutor(executionRequest);
+};
+
 export const createSpellsourceSchema = async () =>
   wrapSchema({
-    schema: await schemaFromExecutor(executor),
+    schema: await schemaFromExecutor(httpExecutor),
     executor,
   });

spellsource-graphql/src/schema/spellsource.ts

@@ -1,13 +1,18 @@
 package com.hiddenswitch.framework.impl;
 
 import com.hiddenswitch.framework.Accounts;
+import com.hiddenswitch.framework.Matchmaking;
 import com.hiddenswitch.framework.graphql.*;
 import graphql.kickstart.tools.GraphQLSubscriptionResolver;
-import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.MessageConsumer;
 import org.apache.commons.lang3.NotImplementedException;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
 import javax.naming.AuthenticationException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * GraphQL subscription resolver.
@@ -37,9 +42,19 @@ public Publisher<ServerGameMessage> gameMessages() throws Exception {
 		return GraphQLGameBridge.gameMessagesPublisher(userId);
 	}
 
+	/**
+	 * Emits a {@link MatchFound} when the matchmaking system creates a game for the
+	 * authenticated user. Listens on the event bus address
+	 * {@code matchmaking:enqueue:{userId}} which is published to by
+	 * {@link Matchmaking#notifyGameReady(String, String)}.
+	 */
 	@Override
 	public Publisher<MatchFound> matchFound() throws Exception {
-		throw new NotImplementedException();
+		var userId = Accounts.userId();
+		if (userId == null) {
+			throw new AuthenticationException("must be authenticated");
+		}
+		return new MatchFoundPublisher(userId);
 	}
 
 	@Override
@@ -56,4 +71,66 @@ public Publisher<Invite> inviteUpdated() throws Exception {
 	public Publisher<EditableCard> editableCardUpdated() throws Exception {
 		throw new NotImplementedException();
 	}
+
+	/**
+	 * A Publisher that listens on the event bus for matchmaking notifications and emits
+	 * {@link MatchFound} when the user is matched into a game.
+	 * <p>
+	 * The event bus message body is the gameId string. The publisher replies to the
+	 * event bus message to acknowledge receipt (the matchmaker waits for this reply
+	 * before proceeding).
+	 */
+	private static class MatchFoundPublisher implements Publisher<MatchFound> {
+		private final String userId;
+
+		MatchFoundPublisher(String userId) {
+			this.userId = userId;
+		}
+
+		@Override
+		public void subscribe(Subscriber<? super MatchFound> subscriber) {
+			var cancelled = new AtomicBoolean(false);
+			var address = Matchmaking.MATCHMAKING_ENQUEUE + userId;
+			var vertx = Vertx.currentContext().owner();
+			var eventBus = vertx.eventBus();
+
+			MessageConsumer<String> consumer = eventBus.consumer(address);
+
+			subscriber.onSubscribe(new Subscription() {
+				@Override
+				public void request(long n) {
+					// backpressure not needed; match events are rare
+				}
+
+				@Override
+				public void cancel() {
+					cancelled.set(true);
+					consumer.unregister();
+				}
+			});
+
+			consumer.handler(message -> {
+				if (cancelled.get()) return;
+
+				var gameId = message.body();
+				// Reply to acknowledge receipt so the matchmaker knows the player was notified
+				message.reply("ok");
+
+				var matchFound = MatchFound.builder()
+						.setGameId(gameId)
+						.setUrl("")
+						.setPlayerKey("")
+						.setPlayerSecret("")
+						.build();
+
+				subscriber.onNext(matchFound);
+			});
+
+			consumer.exceptionHandler(err -> {
+				if (!cancelled.get()) {
+					subscriber.onError(err);
+				}
+			});
+		}
+	}
 }