adds support for subscriptions

3 files changed, 108 insertions, 0 deletions

diff --git a/source/http.ts b/source/http.ts
index 7c747e1..c4a70c8 100755
--- a/source/http.ts
+++ b/source/http.ts
@@ -54,6 +54,9 @@ export function GraphQLHTTP<
     ...options
   }: GQLOptions<Ctx, Req>): GraphQLHandler<Req> {
   return async (request: Req) => {
+    if (options.subscriptions && request.headers.get("upgrade")?.toLowerCase() === "websocket")
+      return options.subscriptions(request as unknown as Request);
+
     const accept = request.headers.get("Accept") || "";
 
     const typeList = ["application/json", "text/html", "text/plain", "*/*"]
diff --git a/source/utility/types.ts b/source/utility/types.ts
index 0d8b443..2311c25 100755
--- a/source/utility/types.ts
+++ b/source/utility/types.ts
@@ -41,6 +41,12 @@ export interface GQLOptions<Context, Req extends GQLRequest = GQLRequest> extend
   playgroundOptions?: Omit<RenderPageOptions, "endpoint">;
   /** The executable schema to query against. */
   schema: GraphQLSchema;
+  /**
+   * WebSocket subscription handler returned by `GraphQLWS()`. When set, any
+   * incoming request with `Upgrade: websocket` is delegated to it before HTTP
+   * content negotiation runs.
+   */
+  subscriptions?: (request: Request) => Response;
 }
 
 /** A single GraphQL operation — either a `query` or a `mutation`, never both. */
diff --git a/source/ws.ts b/source/ws.ts
new file mode 100644
index 0000000..fdd9fb1
--- /dev/null
+++ b/source/ws.ts
@@ -0,0 +1,99 @@
+
+
+
+/*** IMPORT ------------------------------------------- ***/
+
+import { makeServer, type ServerOptions } from "graphql-ws";
+
+/*** UTILITY ------------------------------------------ ***/
+
+import type { GraphQLSchema } from "graphql";
+
+/*** EXPORT ------------------------------------------- ***/
+
+/**
+ * Configuration accepted by {@link GraphQLWS}.
+ *
+ * Wraps `graphql-ws`'s {@link ServerOptions} and requires a `schema`. Anything
+ * else accepted by `graphql-ws` (custom `execute`, `subscribe`, `onConnect`,
+ * `onSubscribe`, etc.) can be passed through.
+ */
+export type WSOptions<Ctx = unknown> = {
+  /** Builds the context value passed to resolvers. Runs per WS connection. */
+  context?: (ctx: { connectionParams?: Record<string, unknown> }) => Ctx | Promise<Ctx>;
+  /** The executable schema to subscribe against. */
+  schema: GraphQLSchema;
+} & Partial<Omit<ServerOptions, "schema" | "context">>;
+
+/**
+ * Builds a WebSocket subscription handler that speaks the
+ * `graphql-transport-ws` protocol via `graphql-ws`.
+ *
+ * The returned function expects a Fetch `Request` with an `Upgrade: websocket`
+ * header. It performs the upgrade with `Deno.upgradeWebSocket` and hands the
+ * socket to `graphql-ws`. Pass the result as `subscriptions` to {@link GraphQLHTTP}
+ * so HTTP and WS share one endpoint.
+ *
+ * @example
+ * ```ts
+ * import { executeSchema, GraphQLHTTP, GraphQLWS, PubSub, gql } from "@eol/gq";
+ *
+ * const pubsub = new PubSub();
+ *
+ * const schema = executeSchema({
+ *   resolvers: {
+ *     Subscription: {
+ *       pinged: { subscribe: () => pubsub.asyncIterator(["PING"]) }
+ *     }
+ *   },
+ *   typeDefs: gql`type Subscription { pinged: String }`
+ * });
+ *
+ * const subscriptions = GraphQLWS({ schema });
+ * Deno.serve(GraphQLHTTP({ schema, subscriptions }));
+ * ```
+ */
+export function GraphQLWS<Ctx = unknown>(options: WSOptions<Ctx>) {
+  const server = makeServer({
+    ...options,
+    context: options.context as ServerOptions["context"],
+    schema: options.schema
+  });
+
+  return (request: Request): Response => {
+    const { response, socket } = Deno.upgradeWebSocket(request, {
+      protocol: "graphql-transport-ws"
+    });
+
+    socket.onopen = () => {
+      const closed = server.opened(
+        {
+          close: (code, reason) => socket.close(code, reason),
+          onMessage: (cb) => {
+            socket.onmessage = async (event) => {
+              try {
+                await cb(
+                  typeof event.data === "string" ?
+                    event.data :
+                    await (event.data as Blob).text()
+                );
+              } catch (err) {
+                socket.close(1011, (err as Error).message);
+              }
+            };
+          },
+          protocol: socket.protocol,
+          send: (data) => Promise.resolve(socket.send(data))
+        },
+        {
+          request,
+          socket
+        }
+      );
+
+      socket.onclose = (event) => closed(event.code, event.reason);
+    };
+
+    return response;
+  };
+}