From bd3ecbe7ace2df12d023da82685097a0420762bf Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 19:24:00 +0200 Subject: [PATCH 1/8] fix(handlersController): group and dedupe handlers on merge --- .../experimental/handlers-controller.test.ts | 80 ++++++++++++++++++- src/core/experimental/handlers-controller.ts | 33 +++++--- src/core/ws.ts | 22 +++-- 3 files changed, 117 insertions(+), 18 deletions(-) diff --git a/src/core/experimental/handlers-controller.test.ts b/src/core/experimental/handlers-controller.test.ts index 6dd4a2c65..edd33bccd 100644 --- a/src/core/experimental/handlers-controller.test.ts +++ b/src/core/experimental/handlers-controller.test.ts @@ -1,8 +1,61 @@ import { http } from '../http' import { graphql } from '../../graphql' import { ws } from '../ws' -import { getSiblingHandlers } from '../utils/internal/attachSiblingHandlers' -import { InMemoryHandlersController } from './handlers-controller' +import { + attachSiblingHandlers, + getSiblingHandlers, +} from '../utils/internal/attachSiblingHandlers' +import { + groupHandlersByKind, + InMemoryHandlersController, +} from './handlers-controller' + +describe(groupHandlersByKind, () => { + it('groups handlers attached as siblings of siblings', () => { + const grandchildHandler = http.get('/grandchild', () => {}) + const childHandler = attachSiblingHandlers( + http.get('/child', () => {}), + [grandchildHandler], + ) + const ownerHandler = attachSiblingHandlers( + http.get('/owner', () => {}), + [childHandler], + ) + + expect(groupHandlersByKind([ownerHandler]).request).toEqual([ + ownerHandler, + childHandler, + grandchildHandler, + ]) + }) + + it('groups nested siblings of a different kind into their own bucket', () => { + const chat = ws.link('*') + const wsHandler = chat.addEventListener('connection', () => {}) + const [upgradeHandler] = getSiblingHandlers(wsHandler) + const ownerHandler = attachSiblingHandlers( + http.get('/owner', () => {}), + [wsHandler], + ) + + const groups = groupHandlersByKind([ownerHandler]) + + expect(groups.websocket).toEqual([wsHandler]) + expect(groups.request).toEqual([ownerHandler, upgradeHandler]) + }) + + it('does not recurse infinitely given a cyclic sibling graph', () => { + const handlerOne = http.get('/one', () => {}) + const handlerTwo = http.get('/two', () => {}) + attachSiblingHandlers(handlerOne, [handlerTwo]) + attachSiblingHandlers(handlerTwo, [handlerOne]) + + expect(groupHandlersByKind([handlerOne]).request).toEqual([ + handlerOne, + handlerTwo, + ]) + }) +}) describe('constructor', () => { it('places the sibling in its own kind bucket', () => { @@ -149,6 +202,29 @@ describe(InMemoryHandlersController.prototype.use, () => { expect(controller.getHandlersByKind('websocket')).toEqual([wsOne, wsTwo]) expect(controller.getHandlersByKind('request')).toEqual([upgradeHandler]) }) + + it('dedupes the shared upgrade sibling against already-registered handlers', () => { + const chat = ws.link('*') + const wsOne = chat.addEventListener('connection', () => {}) + const [upgradeHandler] = getSiblingHandlers(wsOne) + const controller = new InMemoryHandlersController([wsOne]) + + const wsTwo = chat.addEventListener('connection', () => {}) + controller.use([wsTwo]) + + expect(controller.getHandlersByKind('websocket')).toEqual([wsTwo, wsOne]) + expect(controller.getHandlersByKind('request')).toEqual([upgradeHandler]) + }) + + it('moves an already-registered handler to the front when used again', () => { + const httpOne = http.get('/one', () => {}) + const httpTwo = http.get('/two', () => {}) + const controller = new InMemoryHandlersController([httpOne, httpTwo]) + + controller.use([httpTwo]) + + expect(controller.getHandlersByKind('request')).toEqual([httpTwo, httpOne]) + }) }) describe(InMemoryHandlersController.prototype.reset, () => { diff --git a/src/core/experimental/handlers-controller.ts b/src/core/experimental/handlers-controller.ts index d016d49ff..d400eba60 100644 --- a/src/core/experimental/handlers-controller.ts +++ b/src/core/experimental/handlers-controller.ts @@ -9,12 +9,21 @@ export type HandlersMap = Partial>> export function groupHandlersByKind(handlers: Array): HandlersMap { const groups: HandlersMap = {} + const visitedHandlers = new Set() - const pushUnique = (kind: AnyHandler['kind'], handler: AnyHandler) => { - const bucket = (groups[kind] ||= []) + const visit = (handler: AnyHandler) => { + if (visitedHandlers.has(handler)) { + return + } - if (!bucket.includes(handler)) { - bucket.push(handler) + visitedHandlers.add(handler) + const bucket = (groups[handler.kind] ||= []) + bucket.push(handler) + + // Recurse so siblings of siblings (user-composed handler + // graphs) are grouped as well, not silently dropped. + for (const sibling of getSiblingHandlers(handler)) { + visit(sibling) } } @@ -22,11 +31,7 @@ export function groupHandlersByKind(handlers: Array): HandlersMap { * @note `Object.groupBy` is not implemented in Node.js v20. */ for (const handler of handlers) { - pushUnique(handler.kind, handler) - - for (const sibling of getSiblingHandlers(handler)) { - pushUnique(sibling.kind, sibling) - } + visit(handler) } return groups @@ -86,11 +91,19 @@ export abstract class HandlersController { // Prepend overrides to their respective kind buckets so they take // priority over existing handlers while preserving input order. + // Drop existing references that reappear in the overrides (e.g. a + // shared upgrade sibling from the same link) so a handler is never + // registered twice. for (const kind in overrides) { const overridesForKind = overrides[kind as AnyHandler['kind']]! const existingForKind = handlers[kind as AnyHandler['kind']] handlers[kind as AnyHandler['kind']] = existingForKind - ? [...overridesForKind, ...existingForKind] + ? [ + ...overridesForKind, + ...existingForKind.filter((existingHandler) => { + return !overridesForKind.includes(existingHandler) + }), + ] : overridesForKind } diff --git a/src/core/ws.ts b/src/core/ws.ts index 0c99ee823..a29990b88 100644 --- a/src/core/ws.ts +++ b/src/core/ws.ts @@ -88,6 +88,21 @@ export type WebSocketLink = { ) => void } +/** + * Creates a request handler that responds to WebSocket upgrade + * requests whose URL matches the given path. + * + * @internal + */ +export function createWebSocketUpgradeHandler(url: Path) { + return http.get(({ request }) => { + return ( + request.headers.get('upgrade')?.toLowerCase() === 'websocket' && + matchRequestUrl(new URL(resolveWebSocketUrl(request.url)), url).matches + ) + }, ws.onUpgrade) +} + /** * Intercepts outgoing WebSocket connections to the given URL. * @@ -112,12 +127,7 @@ function createWebSocketLinkHandler(url: Path): WebSocketLink { // WebSocketHandler returned by this link. `groupHandlersByKind` dedupes // by reference, so it lands in the `request` bucket exactly once regardless // of which subset of WS handlers the user ends up registering. - const upgradeHandler = http.get(({ request }) => { - return ( - request.headers.get('upgrade')?.toLowerCase() === 'websocket' && - matchRequestUrl(new URL(resolveWebSocketUrl(request.url)), url).matches - ) - }, ws.onUpgrade) + const upgradeHandler = createWebSocketUpgradeHandler(url) return { get clients() { From 78bf034ce21a08c5a539e11c456f3d224f5b89d4 Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 20:26:08 +0200 Subject: [PATCH 2/8] feat: support graphql subscriptions --- package.json | 2 + pnpm-lock.yaml | 244 ++++ src/graphql/graphql-handler.ts | 58 +- src/graphql/graphql-subscription.test.ts | 86 ++ src/graphql/graphql-subscription.ts | 1102 +++++++++++++++++ src/graphql/graphql.ts | 16 + src/graphql/index.ts | 15 + .../graphql-api/graphql-subscription.test.ts | 785 ++++++++++++ 8 files changed, 2285 insertions(+), 23 deletions(-) create mode 100644 src/graphql/graphql-subscription.test.ts create mode 100644 src/graphql/graphql-subscription.ts create mode 100644 test/node/graphql-api/graphql-subscription.test.ts diff --git a/package.json b/package.json index d838dc892..ccbd66260 100644 --- a/package.json +++ b/package.json @@ -292,6 +292,8 @@ "fs-teardown": "^0.3.0", "glob": "^13.0.6", "graphql": "^16.13.2", + "graphql-ws": "^5.16.0", + "graphql-yoga": "^5.13.4", "jsdom": "^25.0.1", "json-bigint": "^1.0.0", "knip": "^6.4.1", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9104a9402..cf5f2c170 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -138,6 +138,12 @@ importers: graphql: specifier: ^16.13.2 version: 16.14.2 + graphql-ws: + specifier: ^5.16.0 + version: 5.16.2(graphql@16.14.2) + graphql-yoga: + specifier: ^5.13.4 + version: 5.21.2(graphql@16.14.2) jsdom: specifier: ^25.0.1 version: 25.0.1 @@ -433,6 +439,18 @@ packages: '@emnapi/wasi-threads@1.2.2': resolution: {integrity: sha512-c95qOXkHdydNKhscBTebqEC1CVAZpyqOfVfBzQ1qgzyl3gfeldUjIggDbIZgDKsHLgnsM+igH7TJ/eAasaVuMA==} + '@envelop/core@5.5.1': + resolution: {integrity: sha512-3DQg8sFskDo386TkL5j12jyRAdip/8yzK3x7YGbZBgobZ4aKXrvDU0GppU0SnmrpQnNaiTUsxBs9LKkwQ/eyvw==} + engines: {node: '>=18.0.0'} + + '@envelop/instrumentation@1.0.0': + resolution: {integrity: sha512-cxgkB66RQB95H3X27jlnxCRNTmPuSTgmBAq6/4n2Dtv4hsk4yz8FadA1ggmd0uZzvKqWD6CR+WFgTjhDqg7eyw==} + engines: {node: '>=18.0.0'} + + '@envelop/types@5.2.1': + resolution: {integrity: sha512-CsFmA3u3c2QoLDTfEpGr4t25fjMU31nyvse7IzWTvb0ZycuPjMjb0fjlheh+PbhBYb9YLugnT2uY6Mwcg1o+Zg==} + engines: {node: '>=18.0.0'} + '@epic-web/invariant@1.0.0': resolution: {integrity: sha512-lrTPqgvfFQtR/eY/qkIzp98OGdNJu0m5ji3q/nJI8v3SXkRKEnWiOxMmbvcSoAIzv/cGiuvRy57k4suKQSAdwA==} @@ -599,6 +617,9 @@ packages: '@fastify/ajv-compiler@4.0.5': resolution: {integrity: sha512-KoWKW+MhvfTRWL4qrhUwAAZoaChluo0m0vbiJlGMt2GXvL4LVPQEjt8kSpHI3IBq5Rez8fg+XeH3cneztq+C7A==} + '@fastify/busboy@3.2.0': + resolution: {integrity: sha512-m9FVDXU3GT2ITSe0UaMA5rU3QkfC/UXtCU8y0gSN/GugTqtVldOBWIB5V6V3sbmenVZUIpU6f+mPEO2+m5iTaA==} + '@fastify/error@4.2.0': resolution: {integrity: sha512-RSo3sVDXfHskiBZKBPRgnQTtIqpi/7zhJOEmAxCiBcM7d0uwdGdxLlsCaLzGs8v8NnxIRlfG0N51p5yFaOentQ==} @@ -617,11 +638,53 @@ packages: '@fastify/websocket@11.2.0': resolution: {integrity: sha512-3HrDPbAG1CzUCqnslgJxppvzaAZffieOVbLp1DAy1huCSynUWPifSvfdEDUR8HlJLp3sp1A36uOM2tJogADS8w==} + '@graphql-tools/executor@1.5.5': + resolution: {integrity: sha512-JAdn4G+ehthnGdJABS+wO6LxAIcoGPiSRHIz1ECYLtaQGkpFIdqH6BLUbZcToX/W/nmOG0kTFLoepCGkugbsBA==} + engines: {node: '>=16.0.0'} + peerDependencies: + graphql: ^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0 + + '@graphql-tools/merge@9.2.0': + resolution: {integrity: sha512-kChDH/sOxm3TCICup5NgTiz/aJBk+5GAuC0lfJS8FcRfsqZvlCkNwpsYTGAATBCJMrgZ9+csRugCjS97jPcNMw==} + engines: {node: '>=16.0.0'} + peerDependencies: + graphql: ^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0 + + '@graphql-tools/schema@10.0.36': + resolution: {integrity: sha512-g8S5aLirZInoi3NojzmBxwsZfVawOE0UBlVWYe8kDAR0FxS0riBDiyW7JnlAKayooHMRAMwGaze4RQU3VTTyig==} + engines: {node: '>=16.0.0'} + peerDependencies: + graphql: ^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0 + + '@graphql-tools/utils@10.11.0': + resolution: {integrity: sha512-iBFR9GXIs0gCD+yc3hoNswViL1O5josI33dUqiNStFI/MHLCEPduasceAcazRH77YONKNiviHBV8f7OgcT4o2Q==} + engines: {node: '>=16.0.0'} + peerDependencies: + graphql: ^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0 + + '@graphql-tools/utils@11.2.0': + resolution: {integrity: sha512-eu9h1R3j/wWc4rvmYJF5AKtlwniDzstrZ/c6KSz+HdI+n7I7iog9xyKmBfpUwSbG1TqPNZBzWjFMkzdYOKq6Bg==} + engines: {node: '>=16.0.0'} + peerDependencies: + graphql: ^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0 + '@graphql-typed-document-node/core@3.2.0': resolution: {integrity: sha512-mB9oAsNCm9aM3/SOv4YtBMqZbYj10R7dkq8byBqxGY/ncFwhf2oQzMV+LCRlWoDSEBJ3COiR1yeDvMtsoOsuFQ==} peerDependencies: graphql: ^0.8.0 || ^0.9.0 || ^0.10.0 || ^0.11.0 || ^0.12.0 || ^0.13.0 || ^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0 + '@graphql-yoga/logger@2.0.1': + resolution: {integrity: sha512-Nv0BoDGLMg9QBKy9cIswQ3/6aKaKjlTh87x3GiBg2Z4RrjyrM48DvOOK0pJh1C1At+b0mUIM67cwZcFTDLN4sA==} + engines: {node: '>=18.0.0'} + + '@graphql-yoga/subscription@5.0.5': + resolution: {integrity: sha512-oCMWOqFs6QV96/NZRt/ZhTQvzjkGB4YohBOpKM4jH/lDT4qb7Lex/aGCxpi/JD9njw3zBBtMqxbaC22+tFHVvw==} + engines: {node: '>=18.0.0'} + + '@graphql-yoga/typed-event-target@3.0.2': + resolution: {integrity: sha512-ZpJxMqB+Qfe3rp6uszCQoag4nSw42icURnBRfFYSOmTgEeOe4rD0vYlbA8spvCu2TlCesNTlEN9BLWtQqLxabA==} + engines: {node: '>=18.0.0'} + '@hono/node-server@1.19.14': resolution: {integrity: sha512-GwtvgtXxnWsucXvbQXkRgqksiH2Qed37H9xHZocE5sA3N8O8O8/8FA3uclQXxXVzc9XBZuEOMK7+r02FmSpHtw==} engines: {node: '>=18.14.1'} @@ -1092,6 +1155,9 @@ packages: '@quansync/fs@1.0.0': resolution: {integrity: sha512-4TJ3DFtlf1L5LDMaM6CanJ/0lckGNtJcMjQ1NAV6zDmA0tEHKZtxNKin8EgPaVX1YzljbxckyT2tJrpQKAtngQ==} + '@repeaterjs/repeater@3.1.0': + resolution: {integrity: sha512-TaoVksZRSx2KWYYpyLQtMQXXeS98VsgZImzW65xmiVgbYhXLk+aEsmzPLirqVuE4/XuUapH2iMtxUzaBNDzdSQ==} + '@rolldown/binding-android-arm64@1.1.4': resolution: {integrity: sha512-EZLpf/8y7GXkkra90ML47kzik/GMP3EMcE9bPyHmRfxLC6z9+aW5A8poCsoxjrT5GfEcNAAvWwUHjvP1pUQkfw==} engines: {node: ^20.19.0 || >=22.12.0} @@ -1584,6 +1650,30 @@ packages: '@webassemblyjs/wast-printer@1.14.1': resolution: {integrity: sha512-kPSSXE6De1XOR820C90RIo2ogvZG+c3KiHzqUoO/F34Y2shGzesfqv7o57xrxovZJH/MetF5UjroJ/R/3isoiw==} + '@whatwg-node/disposablestack@0.0.6': + resolution: {integrity: sha512-LOtTn+JgJvX8WfBVJtF08TGrdjuFzGJc4mkP8EdDI8ADbvO7kiexYep1o8dwnt0okb0jYclCDXF13xU7Ge4zSw==} + engines: {node: '>=18.0.0'} + + '@whatwg-node/events@0.1.2': + resolution: {integrity: sha512-ApcWxkrs1WmEMS2CaLLFUEem/49erT3sxIVjpzU5f6zmVcnijtDSrhoK2zVobOIikZJdH63jdAXOrvjf6eOUNQ==} + engines: {node: '>=18.0.0'} + + '@whatwg-node/fetch@0.10.13': + resolution: {integrity: sha512-b4PhJ+zYj4357zwk4TTuF2nEe0vVtOrwdsrNo5hL+u1ojXNhh1FgJ6pg1jzDlwlT4oBdzfSwaBwMCtFCsIWg8Q==} + engines: {node: '>=18.0.0'} + + '@whatwg-node/node-fetch@0.8.6': + resolution: {integrity: sha512-BDMdYFcerLQkwA2RTldxOqRCs6ZQD1S7UgP3pUdGUkcbgTrP/V5ko77ZkCww9DHmC4lpoYuwigGfQYj285gMvA==} + engines: {node: '>=18.0.0'} + + '@whatwg-node/promise-helpers@1.3.2': + resolution: {integrity: sha512-Nst5JdK47VIl9UcGwtv2Rcgyn5lWtZ0/mhRQ4G8NN2isxpq2TO30iqHzmwoJycjWuyUfg3GFXqP/gFHXeV57IA==} + engines: {node: '>=16.0.0'} + + '@whatwg-node/server@0.11.0': + resolution: {integrity: sha512-VSdkwnJRr8Yv9UgB2aXB3VUPWwd6Oqnn0hycFwhg9pZgWxJXb7JmhsiXe9tmpMwjHFxli12PGcz9aI63YYloGQ==} + engines: {node: '>=18.0.0'} + '@xtuc/ieee754@1.2.0': resolution: {integrity: sha512-DX8nKgqcGwsc0eJSqYt5lwP4DH5FlHnmuWWBRy7X0NcaGR0ZtuyeESgMwTYVEtxmsNGY+qit4QYT/MIYTOTPeA==} @@ -2177,6 +2267,10 @@ packages: cross-fetch@4.1.0: resolution: {integrity: sha512-uKm5PU+MHTootlWEY+mZ4vvXoCn4fLQxT9dSc1sXVMSFkINTJVN8cAQROpwcKm8bJ/c7rgZVIBWzH5T78sNZZw==} + cross-inspect@1.0.1: + resolution: {integrity: sha512-Pcw1JTvZLSJH83iiGWt6fRcT+BjZlCDRVwYLbUcHzv/CRpB7r0MlSrGbIyQvVSNyGnbt7G4AXuyCiDR3POvZ1A==} + engines: {node: '>=16.0.0'} + cross-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -2758,6 +2852,18 @@ packages: graceful-fs@4.2.11: resolution: {integrity: sha512-RbJ5/jmFcNNCcDV5o9eTnBLJ/HszWV0P73bc+Ff4nS/rJj+YaS6IGyiOL0VoBYX+l1Wrl3k63h/KrH+nhJ0XvQ==} + graphql-ws@5.16.2: + resolution: {integrity: sha512-E1uccsZxt/96jH/OwmLPuXMACILs76pKF2i3W861LpKBCYtGIyPQGtWLuBLkND4ox1KHns70e83PS4te50nvPQ==} + engines: {node: '>=10'} + peerDependencies: + graphql: '>=0.11 <=16' + + graphql-yoga@5.21.2: + resolution: {integrity: sha512-IIRF/3xtjj2D6caAWL9177hQ8tV3mWB3hve1GRnz7njPhQ3iY1jFtSp98fNGv0yV9kaPh9kKQ8JWdJZnedVmDw==} + engines: {node: '>=18.0.0'} + peerDependencies: + graphql: ^15.2.0 || ^16.0.0 + graphql@16.14.2: resolution: {integrity: sha512-Chq1s4CY7jmh8gO2qvLIJyfCDIN+EHLFW/9iShnp1z8FjBQMoodWP1kDC36VAMXXIvAjj4ARa7ntfAV2BrjsbA==} engines: {node: ^12.22.0 || ^14.16.0 || ^16.0.0 || >=17.0.0} @@ -4500,6 +4606,9 @@ packages: file-loader: optional: true + urlpattern-polyfill@10.1.0: + resolution: {integrity: sha512-IGjKp/o0NL3Bso1PymYURCJxMPNAf/ILOpendP9f5B6e1rTJgdgiOvgfoT8VxCAdY+Wisb9uhGaJJf3yZ2V9nw==} + util-deprecate@1.0.2: resolution: {integrity: sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==} @@ -5129,6 +5238,23 @@ snapshots: tslib: 2.8.1 optional: true + '@envelop/core@5.5.1': + dependencies: + '@envelop/instrumentation': 1.0.0 + '@envelop/types': 5.2.1 + '@whatwg-node/promise-helpers': 1.3.2 + tslib: 2.8.1 + + '@envelop/instrumentation@1.0.0': + dependencies: + '@whatwg-node/promise-helpers': 1.3.2 + tslib: 2.8.1 + + '@envelop/types@5.2.1': + dependencies: + '@whatwg-node/promise-helpers': 1.3.2 + tslib: 2.8.1 + '@epic-web/invariant@1.0.0': {} '@epic-web/test-server@0.1.6': @@ -5227,6 +5353,8 @@ snapshots: ajv-formats: 3.0.1(ajv@8.20.0) fast-uri: 3.1.3 + '@fastify/busboy@3.2.0': {} + '@fastify/error@4.2.0': {} '@fastify/fast-json-stringify-compiler@5.1.0': @@ -5253,10 +5381,65 @@ snapshots: - bufferutil - utf-8-validate + '@graphql-tools/executor@1.5.5(graphql@16.14.2)': + dependencies: + '@graphql-tools/utils': 11.2.0(graphql@16.14.2) + '@graphql-typed-document-node/core': 3.2.0(graphql@16.14.2) + '@repeaterjs/repeater': 3.1.0 + '@whatwg-node/disposablestack': 0.0.6 + '@whatwg-node/promise-helpers': 1.3.2 + graphql: 16.14.2 + tslib: 2.8.1 + + '@graphql-tools/merge@9.2.0(graphql@16.14.2)': + dependencies: + '@graphql-tools/utils': 11.2.0(graphql@16.14.2) + graphql: 16.14.2 + tslib: 2.8.1 + + '@graphql-tools/schema@10.0.36(graphql@16.14.2)': + dependencies: + '@graphql-tools/merge': 9.2.0(graphql@16.14.2) + '@graphql-tools/utils': 11.2.0(graphql@16.14.2) + graphql: 16.14.2 + tslib: 2.8.1 + + '@graphql-tools/utils@10.11.0(graphql@16.14.2)': + dependencies: + '@graphql-typed-document-node/core': 3.2.0(graphql@16.14.2) + '@whatwg-node/promise-helpers': 1.3.2 + cross-inspect: 1.0.1 + graphql: 16.14.2 + tslib: 2.8.1 + + '@graphql-tools/utils@11.2.0(graphql@16.14.2)': + dependencies: + '@graphql-typed-document-node/core': 3.2.0(graphql@16.14.2) + '@whatwg-node/promise-helpers': 1.3.2 + cross-inspect: 1.0.1 + graphql: 16.14.2 + tslib: 2.8.1 + '@graphql-typed-document-node/core@3.2.0(graphql@16.14.2)': dependencies: graphql: 16.14.2 + '@graphql-yoga/logger@2.0.1': + dependencies: + tslib: 2.8.1 + + '@graphql-yoga/subscription@5.0.5': + dependencies: + '@graphql-yoga/typed-event-target': 3.0.2 + '@repeaterjs/repeater': 3.1.0 + '@whatwg-node/events': 0.1.2 + tslib: 2.8.1 + + '@graphql-yoga/typed-event-target@3.0.2': + dependencies: + '@repeaterjs/repeater': 3.1.0 + tslib: 2.8.1 + '@hono/node-server@1.19.14(hono@4.12.28)': dependencies: hono: 4.12.28 @@ -5613,6 +5796,8 @@ snapshots: dependencies: quansync: 1.0.0 + '@repeaterjs/repeater@3.1.0': {} + '@rolldown/binding-android-arm64@1.1.4': optional: true @@ -6128,6 +6313,39 @@ snapshots: '@webassemblyjs/ast': 1.14.1 '@xtuc/long': 4.2.2 + '@whatwg-node/disposablestack@0.0.6': + dependencies: + '@whatwg-node/promise-helpers': 1.3.2 + tslib: 2.8.1 + + '@whatwg-node/events@0.1.2': + dependencies: + tslib: 2.8.1 + + '@whatwg-node/fetch@0.10.13': + dependencies: + '@whatwg-node/node-fetch': 0.8.6 + urlpattern-polyfill: 10.1.0 + + '@whatwg-node/node-fetch@0.8.6': + dependencies: + '@fastify/busboy': 3.2.0 + '@whatwg-node/disposablestack': 0.0.6 + '@whatwg-node/promise-helpers': 1.3.2 + tslib: 2.8.1 + + '@whatwg-node/promise-helpers@1.3.2': + dependencies: + tslib: 2.8.1 + + '@whatwg-node/server@0.11.0': + dependencies: + '@envelop/instrumentation': 1.0.0 + '@whatwg-node/disposablestack': 0.0.6 + '@whatwg-node/fetch': 0.10.13 + '@whatwg-node/promise-helpers': 1.3.2 + tslib: 2.8.1 + '@xtuc/ieee754@1.2.0': {} '@xtuc/long@4.2.2': {} @@ -6753,6 +6971,10 @@ snapshots: transitivePeerDependencies: - encoding + cross-inspect@1.0.1: + dependencies: + tslib: 2.8.1 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -7498,6 +7720,26 @@ snapshots: graceful-fs@4.2.11: {} + graphql-ws@5.16.2(graphql@16.14.2): + dependencies: + graphql: 16.14.2 + + graphql-yoga@5.21.2(graphql@16.14.2): + dependencies: + '@envelop/core': 5.5.1 + '@envelop/instrumentation': 1.0.0 + '@graphql-tools/executor': 1.5.5(graphql@16.14.2) + '@graphql-tools/schema': 10.0.36(graphql@16.14.2) + '@graphql-tools/utils': 10.11.0(graphql@16.14.2) + '@graphql-yoga/logger': 2.0.1 + '@graphql-yoga/subscription': 5.0.5 + '@whatwg-node/fetch': 0.10.13 + '@whatwg-node/promise-helpers': 1.3.2 + '@whatwg-node/server': 0.11.0 + graphql: 16.14.2 + lru-cache: 10.4.3 + tslib: 2.8.1 + graphql@16.14.2: {} has-bigints@1.1.0: {} @@ -9369,6 +9611,8 @@ snapshots: schema-utils: 3.3.0 webpack: 5.108.4(esbuild@0.28.1) + urlpattern-polyfill@10.1.0: {} + util-deprecate@1.0.2: {} util.promisify@1.1.3: diff --git a/src/graphql/graphql-handler.ts b/src/graphql/graphql-handler.ts index a8c192a3e..7012c3e99 100644 --- a/src/graphql/graphql-handler.ts +++ b/src/graphql/graphql-handler.ts @@ -184,35 +184,49 @@ export class GraphQLHandler extends RequestHandler< return predicate } - constructor( - operationType: GraphQLOperationType, - predicate: GraphQLPredicate, - endpoint: Path, - resolver: ResponseResolver, any, any>, - options?: RequestHandlerOptions, - ) { + /** + * Creates a GraphQL handler information object from the given + * operation predicate. Normalizes `DocumentNode` and typed document + * string predicates to plain operation names. + */ + static parseGraphQLRequestInfo(args: { + operationType: GraphQLOperationType + predicate: GraphQLPredicate + url: Path + }): GraphQLHandlerInfo { const operationName = GraphQLHandler.#parseOperationName( - predicate, - operationType, + args.predicate, + args.operationType, ) const displayOperationName = typeof operationName === 'function' ? '[custom predicate]' : operationName const header = - operationType === 'all' - ? `${operationType} (origin: ${endpoint.toString()})` - : `${operationType}${displayOperationName ? ` ${displayOperationName}` : ''} (origin: ${endpoint.toString()})` + args.operationType === 'all' + ? `${args.operationType} (origin: ${args.url.toString()})` + : `${args.operationType}${displayOperationName ? ` ${displayOperationName}` : ''} (origin: ${args.url.toString()})` + return { + header, + operationType: args.operationType, + operationName, + } + } + + constructor( + operationType: GraphQLOperationType, + predicate: GraphQLPredicate, + endpoint: Path, + resolver: ResponseResolver, any, any>, + options?: RequestHandlerOptions, + ) { super({ - info: { - header, + info: GraphQLHandler.parseGraphQLRequestInfo({ operationType, - operationName: GraphQLHandler.#parseOperationName( - predicate, - operationType, - ), - }, + predicate, + url: endpoint, + }), resolver, options, }) @@ -242,10 +256,8 @@ export class GraphQLHandler extends RequestHandler< } async parse(args: { request: Request }): Promise { - /** - * If the request doesn't match a specified endpoint, there's no - * need to parse it since there's no case where we would handle this - */ + // If the request doesn't match a specified endpoint, there's no + // need to parse it since there's no case where we would handle this const match = matchRequestUrl(new URL(args.request.url), this.endpoint) const cookies = getAllRequestCookies(args.request) diff --git a/src/graphql/graphql-subscription.test.ts b/src/graphql/graphql-subscription.test.ts new file mode 100644 index 000000000..beb0300fa --- /dev/null +++ b/src/graphql/graphql-subscription.test.ts @@ -0,0 +1,86 @@ +// @vitest-environment node +import { OperationTypeNode, parse } from 'graphql' +import { getSiblingHandlers } from '#core/utils/internal/attachSiblingHandlers' +import type { GraphQLHandlerInfo } from './graphql-handler' +import { + createGraphQLSubscriptionHandler, + GraphQLSubscriptionTransportHandler, +} from './graphql-subscription' + +const subscription = createGraphQLSubscriptionHandler( + 'http://localhost:4000/graphql', +) + +describe('info', () => { + it('resolves handler info for a string operation name', () => { + expect( + subscription('OnCommentAdded', () => {}).info, + ).toEqual({ + header: + 'subscription OnCommentAdded (origin: ws://localhost:4000/graphql)', + operationName: 'OnCommentAdded', + operationType: OperationTypeNode.SUBSCRIPTION, + }) + }) + + it('resolves handler info for a RegExp operation name', () => { + expect(subscription(/Comment/, () => {}).info).toEqual({ + header: 'subscription /Comment/ (origin: ws://localhost:4000/graphql)', + operationName: /Comment/, + operationType: OperationTypeNode.SUBSCRIPTION, + }) + }) + + it('resolves handler info for a DocumentNode operation name', () => { + const node = parse(` + subscription OnCommentAdded { + comment { + id + } + } + `) + + expect(subscription(node, () => {}).info).toEqual({ + header: + 'subscription OnCommentAdded (origin: ws://localhost:4000/graphql)', + operationName: 'OnCommentAdded', + operationType: OperationTypeNode.SUBSCRIPTION, + }) + }) +}) + +describe('predicate', () => { + it('returns true for a matching WebSocket connection url', () => { + const handler = subscription('OnCommentAdded', () => {}) + const url = 'ws://localhost:4000/graphql' + const parsedResult = handler.parse({ url }) + + expect(handler.predicate({ url, parsedResult })).toBe(true) + }) + + it('returns false for a non-matching WebSocket connection url', () => { + const handler = subscription('OnCommentAdded', () => {}) + const url = 'ws://example.com/chat' + const parsedResult = handler.parse({ url }) + + expect(handler.predicate({ url, parsedResult })).toBe(false) + }) +}) + +describe('sibling handlers', () => { + it('shares the transport and upgrade handlers between subscription handlers', () => { + const firstHandler = subscription('OnCommentAdded', () => {}) + const secondHandler = subscription('OnPostAdded', () => {}) + + const firstSiblings = getSiblingHandlers(firstHandler) + const secondSiblings = getSiblingHandlers(secondHandler) + + expect(firstSiblings).toHaveLength(2) + expect(firstSiblings[0]).toBeInstanceOf(GraphQLSubscriptionTransportHandler) + + // The transport and the upgrade handlers must be shared by reference + // so the handlers controller dedupes them across subscription handlers. + expect(secondSiblings[0]).toBe(firstSiblings[0]) + expect(secondSiblings[1]).toBe(firstSiblings[1]) + }) +}) diff --git a/src/graphql/graphql-subscription.ts b/src/graphql/graphql-subscription.ts new file mode 100644 index 000000000..ef03629a0 --- /dev/null +++ b/src/graphql/graphql-subscription.ts @@ -0,0 +1,1102 @@ +import { invariant } from 'outvariant' +import { Emitter, TypedEvent } from 'rettime' +import { parse, OperationTypeNode, type GraphQLError } from 'graphql' +import { resolveWebSocketUrl } from '@mswjs/interceptors' +import type { + WebSocketClientConnectionProtocol, + WebSocketConnectionData, + WebSocketData, + WebSocketServerConnectionProtocol, +} from '@mswjs/interceptors/WebSocket' +import { http } from '#core/http' +import { ws } from '#core/ws' +import { + WebSocketHandler, + kConnect, + type WebSocketHandlerConnection, +} from '#core/handlers/WebSocketHandler' +import { + matchRequestUrl, + type Path, + type PathParams, +} from '#core/utils/matching/matchRequestUrl' +import { attachSiblingHandlers } from '#core/utils/internal/attachSiblingHandlers' +import { jsonParse } from '#core/utils/internal/jsonParse' +import { devUtils } from '#core/utils/internal/devUtils' +import { getTimestamp } from '#core/utils/logging/getTimestamp' +import { toPublicUrl } from '#core/utils/request/toPublicUrl' +import { colors } from '#core/ws/utils/attachWebSocketLogger' +import { + GraphQLHandler, + isDocumentNode, + type DocumentTypeDecoration, + type GraphQLHandlerInfo, + type GraphQLHandlerNameSelector, + type GraphQLQuery, + type GraphQLVariables, +} from './graphql-handler' +import { + parseDocumentNode, + type ParsedGraphQLQuery, +} from './parse-graphql-request' + +/** + * Messages of the `graphql-transport-ws` subprotocol. + * @see https://github.com/graphql/graphql-over-http/blob/main/rfcs/GraphQLOverWebSocket.md + */ +export interface GraphQLWebSocketInitMessage { + type: 'connection_init' + payload?: Record +} + +export interface GraphQLWebSocketSubscribePayload< + Variables extends GraphQLVariables = GraphQLVariables, +> { + operationName?: string | null + query: string + variables?: Variables + extensions?: Record +} + +export interface GraphQLWebSocketSubscribeMessage< + Variables extends GraphQLVariables = GraphQLVariables, +> { + type: 'subscribe' + id: string + payload: GraphQLWebSocketSubscribePayload +} + +export interface GraphQLWebSocketCompleteMessage { + type: 'complete' + id: string +} + +export interface GraphQLWebSocketPingMessage { + type: 'ping' + payload?: Record +} + +export interface GraphQLWebSocketPongMessage { + type: 'pong' + payload?: Record +} + +export type GraphQLWebSocketClientMessage = + | GraphQLWebSocketInitMessage + | GraphQLWebSocketSubscribeMessage + | GraphQLWebSocketCompleteMessage + | GraphQLWebSocketPingMessage + | GraphQLWebSocketPongMessage + +export interface GraphQLWebSocketAcknowledgeMessage { + type: 'connection_ack' + payload?: Record +} + +export interface GraphQLWebSocketNextMessage { + type: 'next' + id: string + payload: GraphQLSubscriptionPayload +} + +export interface GraphQLWebSocketErrorMessage { + type: 'error' + id: string + payload: ReadonlyArray> +} + +export type GraphQLWebSocketServerMessage = + | GraphQLWebSocketAcknowledgeMessage + | GraphQLWebSocketNextMessage + | GraphQLWebSocketErrorMessage + | GraphQLWebSocketCompleteMessage + | GraphQLWebSocketPingMessage + | GraphQLWebSocketPongMessage + +/** + * A GraphQL execution result published to a subscription. + */ +export interface GraphQLSubscriptionPayload< + Query extends GraphQLQuery = GraphQLQuery, +> { + data?: Query | null + errors?: ReadonlyArray> | null + extensions?: Record +} + +function createAcknowledgeMessage(): string { + return JSON.stringify({ + type: 'connection_ack', + } satisfies GraphQLWebSocketAcknowledgeMessage) +} + +function createNextMessage(args: { + id: string + payload: GraphQLSubscriptionPayload +}): string { + return JSON.stringify({ + id: args.id, + type: 'next', + payload: args.payload, + } satisfies GraphQLWebSocketNextMessage) +} + +function createErrorMessage(args: { + id: string + payload: ReadonlyArray> +}): string { + return JSON.stringify({ + id: args.id, + type: 'error', + payload: args.payload, + } satisfies GraphQLWebSocketErrorMessage) +} + +function createCompleteMessage(args: { id: string }): string { + return JSON.stringify({ + id: args.id, + type: 'complete', + } satisfies GraphQLWebSocketCompleteMessage) +} + +function createPongMessage(): string { + return JSON.stringify({ + type: 'pong', + } satisfies GraphQLWebSocketPongMessage) +} + +function parseGraphQLWebSocketMessage( + data: WebSocketData, +): MessageType | undefined { + if (typeof data !== 'string') { + return undefined + } + + const message = jsonParse(data) + + if (!message || typeof message.type !== 'string') { + return undefined + } + + return message +} + +/** + * A subscriber provided by a `GraphQLSubscriptionHandler` for a particular + * WebSocket connection. Returns true if the handler matched the parsed + * subscribe operation and resolved it. + */ +type GraphQLSubscriptionSubscriber = (args: { + node: ParsedGraphQLQuery + message: GraphQLWebSocketSubscribeMessage +}) => boolean + +interface GraphQLSubscriptionConnection { + client: WebSocketClientConnectionProtocol + server: WebSocketServerConnectionProtocol + subscribers: Map + subscriptions: Map + isBound: boolean +} + +/** + * A WebSocket handler implementing the `graphql-transport-ws` protocol + * session for a single GraphQL endpoint. One transport is shared across + * all subscription handlers created from the same `graphql.link()` call + * (attached to each of them as a sibling handler). + * + * The transport owns the protocol/session concerns: connection + * acknowledgement, keep-alive, the per-connection registry of active + * subscriptions, and dispatching parsed `subscribe` operations to the + * matching subscription handler. + */ +export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { + #connections: Map + + constructor(url: Path) { + super(url) + this.#connections = new Map() + } + + /** + * Registers the given handler as a subscriber to the GraphQL + * subscriptions on the given WebSocket connection. Subscribers are + * dispatched in registration order, which follows the handlers + * resolution order (runtime handlers take precedence). + */ + public subscribe( + connection: WebSocketHandlerConnection, + handler: WebSocketHandler, + subscriber: GraphQLSubscriptionSubscriber, + ): void { + const transportConnection = this.#getOrCreateConnection(connection) + transportConnection.subscribers.set(handler, subscriber) + } + + public getConnection( + clientId: string, + ): GraphQLSubscriptionConnection | undefined { + return this.#connections.get(clientId) + } + + /** + * Sends a `next` message with the given payload to the subscription. + */ + public publish(args: { + clientId: string + subscriptionId: string + payload: GraphQLSubscriptionPayload + }): void { + const connection = this.#getConnectionForSubscription({ + clientId: args.clientId, + subscriptionId: args.subscriptionId, + intent: 'publish to', + }) + + if (!connection) { + return + } + + connection.client.send( + createNextMessage({ + id: args.subscriptionId, + payload: args.payload, + }), + ) + } + + /** + * Sends a terminal `error` message to the subscription and + * removes it from the registry of active subscriptions. + */ + public error(args: { + clientId: string + subscriptionId: string + errors: ReadonlyArray> + }): void { + const connection = this.#getConnectionForSubscription({ + clientId: args.clientId, + subscriptionId: args.subscriptionId, + intent: 'error', + }) + + if (!connection) { + return + } + + connection.client.send( + createErrorMessage({ + id: args.subscriptionId, + payload: args.errors, + }), + ) + connection.subscriptions.delete(args.subscriptionId) + } + + /** + * Sends a `complete` message to the subscription and removes it + * from the registry of active subscriptions. + */ + public complete(args: { clientId: string; subscriptionId: string }): void { + const connection = this.#getConnectionForSubscription({ + clientId: args.clientId, + subscriptionId: args.subscriptionId, + intent: 'complete', + }) + + if (!connection) { + return + } + + connection.client.send(createCompleteMessage({ id: args.subscriptionId })) + connection.subscriptions.delete(args.subscriptionId) + } + + /** + * Clears the registry of connections and their active subscriptions. + * @note This method is invoked automatically when the handlers + * controller resets the handlers (e.g. `server.resetHandlers()`). + */ + public reset(): void { + this.#connections.clear() + } + + /** + * @note The transport is the sole owner of logging for GraphQL + * subscription connections. It logs parsed `graphql-transport-ws` + * frames instead of raw WebSocket messages. + */ + public log(connection: WebSocketConnectionData): () => void { + return attachGraphQLSubscriptionLogger(connection) + } + + protected [kConnect](connection: WebSocketHandlerConnection): boolean { + const transportConnection = this.#getOrCreateConnection(connection) + + // Bind the protocol listeners at most once per connection + // (e.g. if this transport gets registered multiple times). + if (transportConnection.isBound) { + return true + } + + transportConnection.isBound = true + const { client } = connection + + client.addEventListener('message', (event) => { + this.#handleClientMessage(client.id, event.data) + }) + + client.addEventListener('close', () => { + this.#connections.delete(client.id) + }) + + return true + } + + #getOrCreateConnection( + connection: WebSocketHandlerConnection, + ): GraphQLSubscriptionConnection { + const existingConnection = this.#connections.get(connection.client.id) + + if (existingConnection) { + return existingConnection + } + + const transportConnection: GraphQLSubscriptionConnection = { + client: connection.client, + server: connection.server, + subscribers: new Map(), + subscriptions: new Map(), + isBound: false, + } + this.#connections.set(connection.client.id, transportConnection) + + return transportConnection + } + + #getConnectionForSubscription(args: { + clientId: string + subscriptionId: string + intent: string + }): GraphQLSubscriptionConnection | undefined { + const connection = this.#connections.get(args.clientId) + + if (!connection || !connection.subscriptions.has(args.subscriptionId)) { + devUtils.warn( + 'Failed to %s the GraphQL subscription "%s": the subscription is no longer active.', + args.intent, + args.subscriptionId, + ) + return undefined + } + + return connection + } + + #handleClientMessage(clientId: string, data: WebSocketData): void { + const connection = this.#connections.get(clientId) + + if (!connection) { + return + } + + const message = + parseGraphQLWebSocketMessage(data) + + if (!message) { + return + } + + switch (message.type) { + case 'connection_init': { + connection.client.send(createAcknowledgeMessage()) + break + } + + case 'ping': { + connection.client.send(createPongMessage()) + break + } + + case 'subscribe': { + this.#handleSubscribeMessage(connection, message) + break + } + + case 'complete': { + connection.subscriptions.delete(message.id) + break + } + } + } + + #handleSubscribeMessage( + connection: GraphQLSubscriptionConnection, + message: GraphQLWebSocketSubscribeMessage, + ): void { + let node: ParsedGraphQLQuery + + try { + node = parseDocumentNode(parse(message.payload.query)) + } catch (error) { + devUtils.warn( + 'Failed to intercept a GraphQL subscription to "%s": the subscription query is not a valid GraphQL document.\n\n%s', + toPublicUrl(connection.client.url), + error, + ) + return + } + + if (node.operationType !== OperationTypeNode.SUBSCRIPTION) { + devUtils.warn( + 'Intercepted a GraphQL %s "%s" over WebSocket: only subscription operations are supported over the WebSocket transport.', + node.operationType, + node.operationName || '(anonymous)', + ) + return + } + + // Register the subscription before dispatching it so the resolver + // can publish to it synchronously. + connection.subscriptions.set(message.id, message) + + for (const subscriber of connection.subscribers.values()) { + if (subscriber({ node, message })) { + return + } + } + + devUtils.warn( + 'Intercepted a GraphQL subscription "%s" to "%s" that has no matching subscription handler. If you wish to mock this subscription, create a subscription handler for it.', + node.operationName || '(anonymous)', + toPublicUrl(connection.client.url), + ) + } +} + +export type GraphQLSubscriptionName< + Query extends GraphQLQuery = GraphQLQuery, + Variables extends GraphQLVariables = GraphQLVariables, +> = GraphQLHandlerNameSelector | DocumentTypeDecoration + +export interface GraphQLSubscriptionResolverInfo< + Query extends GraphQLQuery = GraphQLQuery, + Variables extends GraphQLVariables = GraphQLVariables, +> { + /** + * Path parameters parsed from the WebSocket connection URL. + */ + params: PathParams + + /** + * The name of the intercepted operation. + */ + operationName: string + + /** + * Intercepted GraphQL subscription. + */ + subscription: GraphQLSubscription +} + +export type GraphQLSubscriptionResolver< + Query extends GraphQLQuery = GraphQLQuery, + Variables extends GraphQLVariables = GraphQLVariables, +> = (info: GraphQLSubscriptionResolverInfo) => void + +export interface GraphQLSubscriptionHandlerOptions { + /** + * Mark this handler as used after its first match. + * Used handlers do not match subsequent subscriptions. + */ + once?: boolean +} + +/** + * A WebSocket handler intercepting GraphQL subscriptions by their + * operation name. Matching and resolution are delegated to it by the + * subscription transport (its sibling handler) so the first matching + * handler wins, respecting runtime handler overrides. + */ +export class GraphQLSubscriptionHandler< + Query extends GraphQLQuery = GraphQLQuery, + Variables extends GraphQLVariables = GraphQLVariables, +> extends WebSocketHandler { + public info: GraphQLHandlerInfo + public isUsed: boolean + + readonly #operationName: string | RegExp + readonly #transport: GraphQLSubscriptionTransportHandler + readonly #resolver: GraphQLSubscriptionResolver + readonly #options: GraphQLSubscriptionHandlerOptions + + constructor(args: { + url: Path + operationName: GraphQLSubscriptionName + transport: GraphQLSubscriptionTransportHandler + resolver: GraphQLSubscriptionResolver + options?: GraphQLSubscriptionHandlerOptions + }) { + super(args.url) + + // Create the same GraphQL handler info as request-based GraphQL + // handlers so this handler prints nicely during introspection + // (e.g. `server.listHandlers()`). This also normalizes `DocumentNode` + // and typed document predicates to plain operation names. + this.info = GraphQLHandler.parseGraphQLRequestInfo({ + operationType: OperationTypeNode.SUBSCRIPTION, + predicate: args.operationName, + url: args.url, + }) + + const { operationName } = this.info + + invariant( + typeof operationName !== 'function' && !isDocumentNode(operationName), + 'Failed to create a GraphQL subscription handler: custom predicates are not supported for subscriptions', + ) + + this.#operationName = operationName + this.#transport = args.transport + this.#resolver = args.resolver + this.#options = args.options || {} + this.isUsed = false + } + + public reset(): void { + this.isUsed = false + } + + /** + * @note Individual subscription handlers stay silent. The subscription + * transport owns the GraphQL-aware logging for the entire connection + * (a logger is attached once per matching handler otherwise). + */ + public log(): () => void { + return function detachLogger() {} + } + + protected [kConnect](connection: WebSocketHandlerConnection): boolean { + this.#transport.subscribe(connection, this, (args) => { + return this.#handleSubscribe(connection, args) + }) + + return true + } + + #handleSubscribe( + connection: WebSocketHandlerConnection, + args: { + node: ParsedGraphQLQuery + message: GraphQLWebSocketSubscribeMessage + }, + ): boolean { + if (this.#options.once && this.isUsed) { + return false + } + + const { operationName } = args.node + + if (!operationName || !this.#matchesOperationName(operationName)) { + return false + } + + this.isUsed = true + + const subscription = new GraphQLSubscription({ + message: args.message, + clientId: connection.client.id, + transport: this.#transport, + }) + + this.#resolver({ + params: connection.params, + operationName, + subscription, + }) + + return true + } + + #matchesOperationName(operationName: string): boolean { + if (this.#operationName instanceof RegExp) { + return this.#operationName.test(operationName) + } + + return this.#operationName === operationName + } +} + +/** + * Representation of the intercepted GraphQL subscription. + */ +export class GraphQLSubscription< + Query extends GraphQLQuery = GraphQLQuery, + Variables extends GraphQLVariables = GraphQLVariables, +> { + public id: string + public query: string + public variables: Variables + public extensions?: Record + + readonly #message: GraphQLWebSocketSubscribeMessage + readonly #clientId: string + readonly #transport: GraphQLSubscriptionTransportHandler + + constructor(args: { + message: GraphQLWebSocketSubscribeMessage + clientId: string + transport: GraphQLSubscriptionTransportHandler + }) { + this.id = args.message.id + this.query = args.message.payload.query + this.variables = (args.message.payload.variables || {}) as Variables + this.extensions = args.message.payload.extensions + + this.#message = args.message + this.#clientId = args.clientId + this.#transport = args.transport + } + + /** + * Publish an execution result to the subscribed client. + * + * @example + * subscription.publish({ + * data: { + * postAdded: { + * id: 'abc-123' + * } + * } + * }) + */ + public publish(payload: GraphQLSubscriptionPayload): void { + this.#transport.publish({ + clientId: this.#clientId, + subscriptionId: this.id, + payload, + }) + } + + /** + * Use the given `Iterable` or `AsyncIterable` as the source + * of data for this subscription. Whenever the iterable yields a + * value, it gets published to this subscription. + * + * @example + * subscription.from(async function* () { + * yield { text: 'hello world' } + * }) + */ + public async from( + source: Iterable | AsyncIterable, + ): Promise { + for await (const data of source) { + this.publish({ data }) + } + } + + /** + * Terminate this subscription with the given errors. + * + * @example + * subscription.error([{ message: 'Something went wrong' }]) + */ + public error(errors: ReadonlyArray>): void { + this.#transport.error({ + clientId: this.#clientId, + subscriptionId: this.id, + errors, + }) + } + + /** + * Marks this subscription as complete. + * + * @example + * subscription.complete() + */ + public complete(): void { + this.#transport.complete({ + clientId: this.#clientId, + subscriptionId: this.id, + }) + } + + /** + * Perform this GraphQL subscription as-is. + * This establishes a connection to the actual server, replays + * the intercepted subscription, and forwards the server payloads + * to the GraphQL client. You can intercept, modify, or prevent + * any of the original server messages. + * + * @example + * const postAddedSubscription = subscription.passthrough() + * postAddedSubscription.addEventListener('next', (event) => { + * event.preventDefault() + * event.data.payload.data.postAdded.id = 'mock-id' + * subscription.publish(event.data.payload) + * }) + */ + public passthrough(): GraphQLPassthroughSubscription { + const connection = this.#transport.getConnection(this.#clientId) + + /** + * @note One can only call this method inside the GraphQL subscription + * handler. By that point, the WebSocket connection has been established + * and intercepted so the connection reference is guaranteed. + */ + invariant( + connection, + 'Failed to passthrough the GraphQL subscription ("%s"): the underlying WebSocket connection is closed', + this.query, + ) + + return new GraphQLPassthroughSubscription({ + server: connection.server, + message: this.#message, + }) + } +} + +export type GraphQLPassthroughSubscriptionEventMap = { + connection_ack: TypedEvent + next: TypedEvent + error: TypedEvent + complete: TypedEvent +} + +/** + * Representation of a GraphQL subscription to the actual server. + * You interface with this object from the client's perspective. + */ +export class GraphQLPassthroughSubscription { + readonly #server: WebSocketServerConnectionProtocol + readonly #message: GraphQLWebSocketSubscribeMessage + readonly #emitter: Emitter + readonly #abortController: AbortController + + constructor(args: { + server: WebSocketServerConnectionProtocol + message: GraphQLWebSocketSubscribeMessage + }) { + this.#server = args.server + this.#message = args.message + this.#emitter = new Emitter() + + // An abort controller responsible for removing the server event + // listeners once the subscription is unsubscribed. + this.#abortController = new AbortController() + + this.#server.connect() + this.#server.addEventListener( + 'open', + () => { + // Once the WebSocket server connection is established, send the + // client connection prompt to the server. This lets the server + // connect and authorize this client. + this.#server.send( + JSON.stringify({ + type: 'connection_init', + } satisfies GraphQLWebSocketInitMessage), + ) + }, + { signal: this.#abortController.signal }, + ) + + this.#server.addEventListener( + 'message', + (event) => { + const message = + parseGraphQLWebSocketMessage( + event.data, + ) + + if (!message) { + return + } + + switch (message.type) { + case 'connection_ack': { + // Prevent the original acknowledgement from being forwarded + // to the client: the client has already received a mocked + // acknowledgement upon connecting. + event.preventDefault() + this.#emitter.emit(new TypedEvent('connection_ack')) + + // Once the GraphQL server acknowledges the connection, send + // the subscription intent message. This is the same message + // as was sent from the GraphQL client. + this.#server.send(JSON.stringify(this.#message)) + break + } + + case 'next': { + if (message.id !== this.#message.id) { + break + } + + const nextEvent = new TypedEvent('next', { data: message }) + this.#emitter.emit(nextEvent) + + if (nextEvent.defaultPrevented) { + event.preventDefault() + } + + break + } + + case 'error': { + if (message.id !== this.#message.id) { + break + } + + const errorEvent = new TypedEvent('error', { data: message }) + this.#emitter.emit(errorEvent) + + if (errorEvent.defaultPrevented) { + event.preventDefault() + } + + break + } + + case 'complete': { + if (message.id !== this.#message.id) { + break + } + + const completeEvent = new TypedEvent('complete', { data: message }) + this.#emitter.emit(completeEvent) + + if (completeEvent.defaultPrevented) { + event.preventDefault() + } + + break + } + } + }, + { signal: this.#abortController.signal }, + ) + } + + /** + * Add an event listener to the given GraphQL subscription event. + * + * @example + * const onPostAddedSubscription = subscription.passthrough() + * onPostAddedSubscription.addEventListener('next', (event) => { + * console.log(event.data) + * // { id, payload, ... } + * }) + */ + public addEventListener< + EventType extends keyof GraphQLPassthroughSubscriptionEventMap & string, + >( + event: EventType, + listener: Emitter.Listener< + Emitter, + EventType + >, + ): void { + this.#emitter.on(event, listener, { + signal: this.#abortController.signal, + }) + } + + /** + * Unsubscribe from this passthrough GraphQL subscription. + * This closes the underlying server connection. + * + * @note Unsubscribing from the original subscription has no + * effect on the intercepted `subscription` object. + * + * @example + * const onPostAddedSubscription = subscription.passthrough() + * onPostAddedSubscription.unsubscribe() + */ + public unsubscribe(): void { + this.#abortController.abort() + this.#emitter.removeAllListeners() + this.#server.close() + } +} + +function logGraphQLFrame(args: { + color: string + label: string + payload?: unknown +}): void { + const timestamp = getTimestamp({ milliseconds: true }) + + if (typeof args.payload === 'undefined') { + // eslint-disable-next-line no-console + console.log( + devUtils.formatMessage(`${timestamp} %c${args.label}%c`), + `color:${args.color}`, + 'color:inherit', + ) + return + } + + console.groupCollapsed( + devUtils.formatMessage(`${timestamp} %c${args.label}%c`), + `color:${args.color}`, + 'color:inherit', + ) + // eslint-disable-next-line no-console + console.log(args.payload) + console.groupEnd() +} + +/** + * Attach a GraphQL-aware logger to the intercepted WebSocket connection. + * Unlike the raw WebSocket logger, this logger prints parsed + * `graphql-transport-ws` frames relevant to the subscription. + */ +function attachGraphQLSubscriptionLogger( + connection: WebSocketConnectionData, +): () => void { + const { client } = connection + const abortController = new AbortController() + + logGraphQLFrame({ + color: colors.system, + label: `GraphQL subscription connection ${toPublicUrl(client.url)}`, + }) + + client.addEventListener( + 'message', + (event) => { + const message = + parseGraphQLWebSocketMessage(event.data) + + if (!message) { + return + } + + switch (message.type) { + case 'subscribe': { + logGraphQLFrame({ + color: colors.outgoing, + label: `subscribe (id: ${message.id})`, + payload: message.payload, + }) + break + } + + case 'complete': { + logGraphQLFrame({ + color: colors.outgoing, + label: `complete (id: ${message.id})`, + }) + break + } + } + }, + { signal: abortController.signal }, + ) + + // Proxy `client.send` to log the frames published to the client + // (`client.send` does not dispatch any observable events). + const originalClientSend = client.send + + client.send = new Proxy(client.send, { + apply: (target, thisArg, args) => { + const [data] = args + const message = + parseGraphQLWebSocketMessage(data) + + if (message) { + switch (message.type) { + case 'next': { + logGraphQLFrame({ + color: colors.mocked, + label: `next (id: ${message.id})`, + payload: message.payload, + }) + break + } + + case 'error': { + logGraphQLFrame({ + color: colors.mocked, + label: `error (id: ${message.id})`, + payload: message.payload, + }) + break + } + + case 'complete': { + logGraphQLFrame({ + color: colors.mocked, + label: `complete (id: ${message.id})`, + }) + break + } + } + } + + return Reflect.apply(target, thisArg, args) + }, + }) + + return function detachLogger() { + abortController.abort() + client.send = originalClientSend + } +} + +export type GraphQLSubscriptionHandlerFactory = < + Query extends GraphQLQuery = GraphQLQuery, + Variables extends GraphQLVariables = GraphQLVariables, +>( + operationName: GraphQLSubscriptionName, + resolver: GraphQLSubscriptionResolver, + options?: GraphQLSubscriptionHandlerOptions, +) => GraphQLSubscriptionHandler + +/** + * Create a `subscription()` handler factory bound to the given GraphQL + * endpoint. All subscription handlers created by the factory share a single + * subscription transport and a single WebSocket upgrade handler, both + * attached to each handler as siblings. + * + * @example + * const subscription = createGraphQLSubscriptionHandler('https://api.example.com/graphql') + * subscription('OnPostAdded', ({ subscription }) => { + * subscription.publish({ data: { postAdded: { id: 'abc-123' } } }) + * }) + */ +export function createGraphQLSubscriptionHandler( + url: Path, +): GraphQLSubscriptionHandlerFactory { + const webSocketUrl = + typeof url === 'string' ? url.replace(/^http/, 'ws') : url + + const transport = new GraphQLSubscriptionTransportHandler(webSocketUrl) + + // The `upgrade` request handler enables WebSocket interception in Node.js. + // The same handler instance is shared between all subscription handlers + // of this endpoint (sibling handlers are deduped by reference). + const upgradeHandler = http.get(({ request }) => { + return ( + request.headers.get('upgrade')?.toLowerCase() === 'websocket' && + matchRequestUrl(new URL(resolveWebSocketUrl(request.url)), webSocketUrl) + .matches + ) + }, ws.onUpgrade) + + return (operationName, resolver, options) => { + const handler = new GraphQLSubscriptionHandler({ + url: webSocketUrl, + operationName, + transport, + resolver, + options, + }) + + return attachSiblingHandlers(handler, [transport, upgradeHandler]) + } +} diff --git a/src/graphql/graphql.ts b/src/graphql/graphql.ts index 16df704f9..6ac45fd16 100644 --- a/src/graphql/graphql.ts +++ b/src/graphql/graphql.ts @@ -13,6 +13,10 @@ import { type GraphQLPredicate, } from './graphql-handler' import type { Path } from '#core/utils/matching/matchRequestUrl' +import { + createGraphQLSubscriptionHandler, + type GraphQLSubscriptionHandlerFactory, +} from './graphql-subscription' export type GraphQLRequestHandler = < Query extends GraphQLQuery = GraphQLQuery, @@ -65,6 +69,17 @@ export interface GraphQLLinkHandlers { query: GraphQLRequestHandler mutation: GraphQLRequestHandler operation: GraphQLOperationHandler + /** + * Intercept a GraphQL subscription. + * + * @example + * graphql.subscription('OnPostAdded', ({ subscription }) => { + * subscription.publish({ + * data: { postAdded: { id: 'abc-123' } }, + * }) + * }) + */ + subscription: GraphQLSubscriptionHandlerFactory } /** @@ -131,6 +146,7 @@ export const graphql = { 'mutation' as OperationTypeNode, url, ), + subscription: createGraphQLSubscriptionHandler(url), } }, } diff --git a/src/graphql/index.ts b/src/graphql/index.ts index 36e71c34b..20b95135d 100644 --- a/src/graphql/index.ts +++ b/src/graphql/index.ts @@ -18,3 +18,18 @@ export { } from './graphql-handler' export type { ParsedGraphQLRequest } from './parse-graphql-request' + +export { + createGraphQLSubscriptionHandler, + GraphQLSubscription, + GraphQLSubscriptionHandler, + GraphQLSubscriptionTransportHandler, + GraphQLPassthroughSubscription, + type GraphQLSubscriptionHandlerFactory, + type GraphQLSubscriptionHandlerOptions, + type GraphQLSubscriptionName, + type GraphQLSubscriptionPayload, + type GraphQLSubscriptionResolver, + type GraphQLSubscriptionResolverInfo, + type GraphQLPassthroughSubscriptionEventMap, +} from './graphql-subscription' diff --git a/test/node/graphql-api/graphql-subscription.test.ts b/test/node/graphql-api/graphql-subscription.test.ts new file mode 100644 index 000000000..d72909707 --- /dev/null +++ b/test/node/graphql-api/graphql-subscription.test.ts @@ -0,0 +1,785 @@ +// @vitest-environment node +import { HttpResponse, type PathParams } from 'msw' +import { setupServer } from 'msw/node' +import { + graphql, + type GraphQLSubscription, + type GraphQLSubscriptionResolver, +} from 'msw/graphql' +import { DeferredPromise } from '@open-draft/deferred-promise' +import { createTestHttpServer } from '@epic-web/test-server/http' +import { createWebSocketMiddleware } from '@epic-web/test-server/ws' +import { + createPubSub, + createSchema, + createYoga, + type YogaSchemaDefinition, +} from 'graphql-yoga' +import { createClient } from 'graphql-ws' +import { useServer } from 'graphql-ws/lib/use/ws' +import { gql } from '../../support/graphql' + +const server = setupServer() + +beforeAll(() => { + server.listen() +}) + +afterEach(() => { + server.resetHandlers() + vi.restoreAllMocks() +}) + +afterAll(() => { + server.close() +}) + +it('intercepts and mocks a GraphQL subscription', async () => { + const api = graphql.link('http://localhost:4000/graphql') + + server.use( + api.subscription('OnCommentAdded', ({ subscription }) => { + subscription.publish({ + data: { + commentAdded: { + id: '1', + text: 'Hello world', + }, + }, + }) + }), + ) + + const client = createClient({ + url: 'ws://localhost:4000/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + id + text + } + } + `, + }) + + await expect(subscription.next()).resolves.toEqual({ + done: false, + value: { + data: { + commentAdded: { + id: '1', + text: 'Hello world', + }, + }, + }, + }) +}) + +it('marks a subscription as complete', async () => { + const api = graphql.link('http://localhost:4000/graphql') + + server.use( + api.subscription('OnCommentAdded', ({ subscription }) => { + queueMicrotask(() => { + subscription.complete() + }) + }), + ) + + const client = createClient({ + url: 'ws://localhost:4000/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + await expect(subscription.next()).resolves.toEqual({ + done: true, + value: undefined, + }) +}) + +it('terminates a subscription with errors', async () => { + const api = graphql.link('http://localhost:4000/graphql') + + server.use( + api.subscription('OnCommentAdded', ({ subscription }) => { + queueMicrotask(() => { + subscription.error([{ message: 'Something went wrong' }]) + }) + }), + ) + + const client = createClient({ + url: 'ws://localhost:4000/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + await expect(subscription.next()).rejects.toEqual([ + expect.objectContaining({ message: 'Something went wrong' }), + ]) +}) + +it('exposes path parameters from the WebSocket link', async () => { + const paramsPromise = new DeferredPromise() + const api = graphql.link('https://localhost/:service') + + server.use( + api.subscription('OnCommentAdded', ({ params, subscription }) => { + paramsPromise.resolve(params) + subscription.complete() + }), + ) + + const client = createClient({ + url: 'wss://localhost/user-service', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + await subscription.next() + + await expect(paramsPromise).resolves.toEqual({ + service: 'user-service', + }) +}) + +it('scopes published data to the subscribed client', async () => { + const api = graphql.link('http://localhost:4000/graphql') + + server.use( + api.subscription<{ greeting: string }, { name: string }>( + 'OnGreeting', + ({ subscription }) => { + subscription.publish({ + data: { + greeting: `hello, ${subscription.variables.name}`, + }, + }) + queueMicrotask(() => { + subscription.complete() + }) + }, + ), + ) + + const query = gql` + subscription OnGreeting($name: String!) { + greeting(name: $name) + } + ` + + async function collectMessages(name: string): Promise> { + const client = createClient({ url: 'ws://localhost:4000/graphql' }) + const messages: Array = [] + + for await (const result of client.iterate({ + query, + variables: { name }, + })) { + messages.push(result) + } + + return messages + } + + // Each client must receive only the data published to its + // own subscription (subscription ids are unique per socket, + // not across the clients). + await expect(collectMessages('john')).resolves.toEqual([ + { data: { greeting: 'hello, john' } }, + ]) + await expect(collectMessages('kate')).resolves.toEqual([ + { data: { greeting: 'hello, kate' } }, + ]) +}) + +it('respects handler overrides for the same operation', async () => { + const api = graphql.link('http://localhost:4000/graphql') + const initialResolver = vi.fn() + + server.use(api.subscription('OnCommentAdded', initialResolver)) + server.use( + api.subscription('OnCommentAdded', ({ subscription }) => { + subscription.publish({ + data: { + commentAdded: { text: 'override' }, + }, + }) + }), + ) + + const client = createClient({ + url: 'ws://localhost:4000/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + // Must receive the data from the override handler. + await expect(subscription.next()).resolves.toEqual({ + done: false, + value: { + data: { + commentAdded: { text: 'override' }, + }, + }, + }) + + // The initial handler must not be called (first matching handler wins). + expect(initialResolver).not.toHaveBeenCalled() +}) + +it('supports one-time subscription handlers', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + + const api = graphql.link('http://localhost:4000/graphql') + const resolver = vi.fn(({ subscription }) => { + subscription.publish({ + data: { commentAdded: { text: 'first' } }, + }) + }) + + server.use(api.subscription('OnCommentAdded', resolver, { once: true })) + + const client = createClient({ + url: 'ws://localhost:4000/graphql', + }) + const query = gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + ` + + const firstSubscription = client.iterate({ query }) + await expect(firstSubscription.next()).resolves.toEqual({ + done: false, + value: { + data: { commentAdded: { text: 'first' } }, + }, + }) + + // The second subscription must not match the used handler. + const secondSubscription = client.iterate({ query }) + const secondNext = secondSubscription.next() + + await expect + .poll(() => vi.mocked(console.warn).mock.calls.flat().join('\n')) + .toMatch(/no matching subscription handler/) + + expect(resolver).toHaveBeenCalledTimes(1) + + secondNext.catch(() => {}) + await client.dispose() +}) + +it('warns on a subscription without a matching handler', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + + const api = graphql.link('http://localhost:4000/graphql') + server.use(api.subscription('OnCommentAdded', () => {})) + + const client = createClient({ + url: 'ws://localhost:4000/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnPostAdded { + postAdded { + id + } + } + `, + }) + const pendingNext = subscription.next() + + await expect + .poll(() => vi.mocked(console.warn).mock.calls.flat().join('\n')) + .toMatch( + /Intercepted a GraphQL subscription "OnPostAdded" to "ws:\/\/localhost:4000\/graphql" that has no matching subscription handler/, + ) + + pendingNext.catch(() => {}) + await client.dispose() +}) + +it('warns when publishing to a subscription after the handlers were reset', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + + const api = graphql.link('http://localhost:4000/graphql') + const subscriptionPromise = new DeferredPromise() + + server.use( + api.subscription('OnCommentAdded', ({ subscription }) => { + subscriptionPromise.resolve(subscription) + }), + ) + + const client = createClient({ + url: 'ws://localhost:4000/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + const pendingNext = subscription.next() + + const interceptedSubscription = await subscriptionPromise + server.resetHandlers() + + // Publishing to a stale subscription must be a warning no-op + // instead of writing to the (possibly unrelated) socket. + interceptedSubscription.publish({ + data: { commentAdded: { text: 'stale' } }, + }) + + expect(console.warn).toHaveBeenCalledWith( + expect.stringMatching( + /Failed to publish to the GraphQL subscription ".+": the subscription is no longer active/, + ), + ) + + pendingNext.catch(() => {}) + await client.dispose() +}) + +it('responds to the protocol ping messages', async () => { + const api = graphql.link('http://localhost:4000/graphql') + server.use(api.subscription('OnCommentAdded', () => {})) + + const socket = new WebSocket('ws://localhost:4000/graphql', [ + 'graphql-transport-ws', + ]) + const messages: Array<{ type: string }> = [] + const pongPromise = new DeferredPromise() + + socket.onopen = () => { + socket.send(JSON.stringify({ type: 'connection_init' })) + } + socket.onmessage = (event) => { + const message = JSON.parse(String(event.data)) + messages.push(message) + + if (message.type === 'connection_ack') { + socket.send(JSON.stringify({ type: 'ping' })) + } + + if (message.type === 'pong') { + pongPromise.resolve() + } + } + + await pongPromise + + expect(messages).toEqual([{ type: 'connection_ack' }, { type: 'pong' }]) + socket.close() +}) + +it('subscribes to extraneous pubsubs', async () => { + const pubsub = createPubSub<{ + commentAdded: [{ commentAdded: { text: string } }] + }>() + const subscriptionEstablishedPromise = new DeferredPromise() + + const api = graphql.link('https://localhost/graphql') + server.use( + api.subscription('OnCommentAdded', ({ subscription }) => { + subscription.from(pubsub.subscribe('commentAdded')) + subscriptionEstablishedPromise.resolve() + }), + api.mutation('AddComment', ({ variables }) => { + const { comment } = variables + + pubsub.publish('commentAdded', { + commentAdded: comment, + }) + + return HttpResponse.json({ + data: { comment }, + }) + }), + ) + + const client = createClient({ + url: 'wss://localhost/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + // Await the subscription to be established before publishing to the + // pubsub. Unlike the client, the pubsub does not replay events published + // before the subscription became active (the mutation below may otherwise + // outrace the WebSocket handshake, e.g. on Node.js 24). + await subscriptionEstablishedPromise + + const comment = { text: 'hello world' } + await fetch('https://localhost/graphql', { + method: 'POST', + headers: { + 'content-type': 'application/json', + }, + body: JSON.stringify({ + query: gql` + mutation AddComment($comment: CommentInput!) { + comment { + text + } + } + `, + variables: { comment }, + }), + }) + + await expect(subscription.next()).resolves.toEqual({ + done: false, + value: { + data: { + commentAdded: comment, + }, + }, + }) +}) + +it('combines extraneous and default pubsubs', async () => { + const pubsub = createPubSub<{ + commentAdded: [{ commentAdded: { text: string } }] + }>() + + const api = graphql.link('https://localhost/graphql') + server.use( + api.subscription('OnCommentAdded', ({ subscription }) => { + subscription.publish({ + data: { + commentAdded: { + text: 'manual comment', + }, + }, + }) + + subscription.from(pubsub.subscribe('commentAdded')) + }), + api.mutation('AddComment', ({ variables }) => { + const { comment } = variables + + pubsub.publish('commentAdded', { commentAdded: comment }) + + return HttpResponse.json({ + data: { comment }, + }) + }), + ) + + const client = createClient({ + url: 'wss://localhost/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + // First, must receive the payload from the manual publish. + await expect(subscription.next()).resolves.toEqual({ + done: false, + value: { + data: { + commentAdded: { text: 'manual comment' }, + }, + }, + }) + + const comment = { text: 'comment from mutation' } + await fetch('https://localhost/graphql', { + method: 'POST', + headers: { + 'content-type': 'application/json', + }, + body: JSON.stringify({ + query: gql` + mutation AddComment($comment: CommentInput!) { + comment { + text + } + } + `, + variables: { comment }, + }), + }) + + // Then, must receive the publish from the mutation. + await expect(subscription.next()).resolves.toEqual({ + done: false, + value: { + data: { + commentAdded: comment, + }, + }, + }) +}) + +async function createTestGraphQLServer(options: { + pathname?: string + schema: YogaSchemaDefinition, Record> +}) { + const pathname = options.pathname || '/graphql' + + const yoga = createYoga({ + schema: options.schema, + graphiql: false, + }) + + const testServer = await createTestHttpServer({ + defineRoutes(router) { + router.get('/graphql/*', ({ req }) => { + return yoga.fetch(req.raw) + }) + }, + }) + const wss = createWebSocketMiddleware({ + server: testServer, + pathname: yoga.graphqlEndpoint, + }) + + const disposeOfServer = useServer( + { + execute: (args: any) => args.execute(args), + subscribe: (args: any) => args.subscribe(args), + onSubscribe: async (ctx, params) => { + const { schema, execute, subscribe, contextFactory, parse, validate } = + yoga.getEnveloped({ + ...ctx, + req: ctx.extra.request, + socket: ctx.extra.socket, + params, + }) + + const args = { + schema, + operationName: params.payload.operationName, + document: parse(params.payload.query), + variableValues: params.payload.variables, + contextValue: await contextFactory(), + execute, + subscribe, + } + + const errors = validate(args.schema, args.document) + + if (errors.length) { + return errors + } + + return args + }, + }, + wss.raw, + ) + + return { + async [Symbol.asyncDispose]() { + await Promise.all([ + testServer[Symbol.asyncDispose](), + wss[Symbol.asyncDispose](), + ]) + await disposeOfServer.dispose() + }, + http: { + url() { + return testServer.http.url(pathname) + }, + }, + ws: { + url() { + return wss.ws.url() + }, + }, + } +} + +it('bypasses a subscription', async () => { + await using testServer = await createTestGraphQLServer({ + schema: createSchema({ + typeDefs: gql` + type Comment { + text: String! + } + + type Query { + comments: [Comment!]! + } + + type Subscription { + commentAdded: Comment! + } + `, + resolvers: { + Subscription: { + commentAdded: { + async *subscribe() { + yield { commentAdded: { text: 'hello world' } } + }, + }, + }, + }, + }), + }) + + const api = graphql.link(testServer.http.url().href) + server.use( + api.subscription('OnCommentAdded', async ({ subscription }) => { + const onCommentAddedSubscription = subscription.passthrough() + onCommentAddedSubscription.addEventListener('next', () => { + onCommentAddedSubscription.unsubscribe() + }) + }), + ) + + const client = createClient({ + url: testServer.ws.url().href, + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + // Must receive the payload from the original server. + await expect(subscription.next()).resolves.toEqual({ + value: { + data: { + commentAdded: { + text: 'hello world', + }, + }, + }, + done: false, + }) +}) + +it('augments original server subscription payload', async () => { + await using testServer = await createTestGraphQLServer({ + schema: createSchema({ + typeDefs: gql` + type Comment { + text: String! + } + + type Query { + comments: [Comment!]! + } + + type Subscription { + commentAdded: Comment! + } + `, + resolvers: { + Subscription: { + commentAdded: { + async *subscribe() { + yield { commentAdded: { text: 'hello world' } } + }, + }, + }, + }, + }), + }) + + const api = graphql.link(testServer.http.url().href) + server.use( + api.subscription('OnCommentAdded', async ({ subscription }) => { + const onCommentAddedSubscription = subscription.passthrough() + onCommentAddedSubscription.addEventListener('next', (event) => { + // Prevent the default server-to-client forwarding. + event.preventDefault() + + // Publish the modified payload to the client. + const { payload } = event.data + payload.data = { + commentAdded: { + text: String(payload.data?.commentAdded?.text).toUpperCase(), + }, + } + + subscription.publish(payload) + + onCommentAddedSubscription.unsubscribe() + }) + }), + ) + + const client = createClient({ + url: testServer.ws.url().href, + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded { + commentAdded { + text + } + } + `, + }) + + // Must receive the payload from the original server. + await expect(subscription.next()).resolves.toEqual({ + value: { + data: { + commentAdded: { + text: 'HELLO WORLD', + }, + }, + }, + done: false, + }) +}) From 9d3bb68e1bc89f9463fdb450fcb3ff6d49724539 Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 20:38:18 +0200 Subject: [PATCH 3/8] chore: use `Promise.withResolvers` over `DeferredPromise` in tests --- .../graphql-api/graphql-subscription.test.ts | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/test/node/graphql-api/graphql-subscription.test.ts b/test/node/graphql-api/graphql-subscription.test.ts index d72909707..d3c2d9297 100644 --- a/test/node/graphql-api/graphql-subscription.test.ts +++ b/test/node/graphql-api/graphql-subscription.test.ts @@ -6,7 +6,6 @@ import { type GraphQLSubscription, type GraphQLSubscriptionResolver, } from 'msw/graphql' -import { DeferredPromise } from '@open-draft/deferred-promise' import { createTestHttpServer } from '@epic-web/test-server/http' import { createWebSocketMiddleware } from '@epic-web/test-server/ws' import { @@ -137,7 +136,7 @@ it('terminates a subscription with errors', async () => { }) it('exposes path parameters from the WebSocket link', async () => { - const paramsPromise = new DeferredPromise() + const paramsPromise = Promise.withResolvers() const api = graphql.link('https://localhost/:service') server.use( @@ -162,7 +161,7 @@ it('exposes path parameters from the WebSocket link', async () => { await subscription.next() - await expect(paramsPromise).resolves.toEqual({ + await expect(paramsPromise.promise).resolves.toEqual({ service: 'user-service', }) }) @@ -338,7 +337,7 @@ it('warns when publishing to a subscription after the handlers were reset', asyn vi.spyOn(console, 'warn').mockImplementation(() => {}) const api = graphql.link('http://localhost:4000/graphql') - const subscriptionPromise = new DeferredPromise() + const subscriptionPromise = Promise.withResolvers() server.use( api.subscription('OnCommentAdded', ({ subscription }) => { @@ -360,7 +359,7 @@ it('warns when publishing to a subscription after the handlers were reset', asyn }) const pendingNext = subscription.next() - const interceptedSubscription = await subscriptionPromise + const interceptedSubscription = await subscriptionPromise.promise server.resetHandlers() // Publishing to a stale subscription must be a warning no-op @@ -387,7 +386,7 @@ it('responds to the protocol ping messages', async () => { 'graphql-transport-ws', ]) const messages: Array<{ type: string }> = [] - const pongPromise = new DeferredPromise() + const pongPromise = Promise.withResolvers() socket.onopen = () => { socket.send(JSON.stringify({ type: 'connection_init' })) @@ -405,7 +404,7 @@ it('responds to the protocol ping messages', async () => { } } - await pongPromise + await pongPromise.promise expect(messages).toEqual([{ type: 'connection_ack' }, { type: 'pong' }]) socket.close() @@ -415,7 +414,7 @@ it('subscribes to extraneous pubsubs', async () => { const pubsub = createPubSub<{ commentAdded: [{ commentAdded: { text: string } }] }>() - const subscriptionEstablishedPromise = new DeferredPromise() + const subscriptionEstablishedPromise = Promise.withResolvers() const api = graphql.link('https://localhost/graphql') server.use( @@ -453,7 +452,7 @@ it('subscribes to extraneous pubsubs', async () => { // pubsub. Unlike the client, the pubsub does not replay events published // before the subscription became active (the mutation below may otherwise // outrace the WebSocket handshake, e.g. on Node.js 24). - await subscriptionEstablishedPromise + await subscriptionEstablishedPromise.promise const comment = { text: 'hello world' } await fetch('https://localhost/graphql', { From b05aa353b54a69faf99876cdd5c08fc94c092c03 Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 21:09:12 +0200 Subject: [PATCH 4/8] chore: consistent jsdoc person --- src/graphql/graphql-subscription.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/graphql/graphql-subscription.ts b/src/graphql/graphql-subscription.ts index ef03629a0..4c51cb372 100644 --- a/src/graphql/graphql-subscription.ts +++ b/src/graphql/graphql-subscription.ts @@ -219,7 +219,7 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { } /** - * Registers the given handler as a subscriber to the GraphQL + * Register the given handler as a subscriber to the GraphQL * subscriptions on the given WebSocket connection. Subscribers are * dispatched in registration order, which follows the handlers * resolution order (runtime handlers take precedence). @@ -240,7 +240,7 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { } /** - * Sends a `next` message with the given payload to the subscription. + * Send a `next` message with the given payload to the subscription. */ public publish(args: { clientId: string @@ -266,7 +266,7 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { } /** - * Sends a terminal `error` message to the subscription and + * Send a terminal `error` message to the subscription and * removes it from the registry of active subscriptions. */ public error(args: { @@ -294,7 +294,7 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { } /** - * Sends a `complete` message to the subscription and removes it + * Send a `complete` message to the subscription and removes it * from the registry of active subscriptions. */ public complete(args: { clientId: string; subscriptionId: string }): void { @@ -313,7 +313,7 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { } /** - * Clears the registry of connections and their active subscriptions. + * Clear the registry of connections and their active subscriptions. * @note This method is invoked automatically when the handlers * controller resets the handlers (e.g. `server.resetHandlers()`). */ From a9fc4019060b718ef1019ee2d7d7fb5e92d7b72a Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 21:24:01 +0200 Subject: [PATCH 5/8] fix: emit `graphql:subscription` event --- .../experimental/frames/websocket-frame.ts | 44 +++++++++ src/core/handlers/WebSocketHandler.ts | 17 +++- src/graphql/graphql-subscription.ts | 51 +++++++++++ test/node/graphql-api/events.test.ts | 51 +++++++++++ .../graphql-api/graphql-subscription.test.ts | 90 ++++++++++++++++++- 5 files changed, 248 insertions(+), 5 deletions(-) create mode 100644 test/node/graphql-api/events.test.ts diff --git a/src/core/experimental/frames/websocket-frame.ts b/src/core/experimental/frames/websocket-frame.ts index 475bf3eaf..8154d98d7 100644 --- a/src/core/experimental/frames/websocket-frame.ts +++ b/src/core/experimental/frames/websocket-frame.ts @@ -23,9 +23,47 @@ export interface WebSocketNetworkFrameOptions { export type WebSocketNetworkFrameEventMap = { connection: WebSocketConnectionEvent + 'graphql:subscription': GraphQLSubscriptionEvent unhandledException: UnhandledWebSocketExceptionEvent } +export interface GraphQLSubscriptionEventInit { + operationName: string + query: string + variables: Record + request: Request +} + +/** + * Emitted when a GraphQL subscription is established over an + * intercepted WebSocket connection (i.e. matched by a subscription + * handler and resolved). + * + * @note The event type is declared here, next to the WebSocket frame + * event map, so the life-cycle event emitters derived from this map + * (e.g. `server.events`) are typed correctly. It carries plain data + * only and is emitted exclusively by the `msw/graphql` module — + * the core stays free of the `graphql` dependency. + */ +export class GraphQLSubscriptionEvent extends TypedEvent< + void, + void, + 'graphql:subscription' +> { + public readonly operationName: string + public readonly query: string + public readonly variables: Record + public readonly request: Request + + constructor(init: GraphQLSubscriptionEventInit) { + super('graphql:subscription') + this.operationName = init.operationName + this.query = init.query + this.variables = init.variables + this.request = init.request + } +} + class WebSocketConnectionEvent< DataType extends { url: URL @@ -115,6 +153,12 @@ export abstract class WebSocketNetworkFrame extends NetworkFrame< for (const handler of handlers) { const handlerConnection = await handler.run(connection, { baseUrl: resolutionContext?.baseUrl?.toString(), + /** + * @note Expose an emit-only reference to this frame's events + * so the handlers can emit additional events not covered by + * the frame (e.g. "graphql:subscription"). + */ + events: this.events, /** * @note Do not emit the "connection" event when running the handler. * Use the run only to get the resolved connection object. diff --git a/src/core/handlers/WebSocketHandler.ts b/src/core/handlers/WebSocketHandler.ts index 7e18d9c56..17277fd2e 100644 --- a/src/core/handlers/WebSocketHandler.ts +++ b/src/core/handlers/WebSocketHandler.ts @@ -1,10 +1,16 @@ import { Emitter } from 'strict-event-emitter' +import type { Emitter as NetworkFrameEmitter } from 'rettime' import { createRequestId, resolveWebSocketUrl } from '@mswjs/interceptors' import type { WebSocketClientConnectionProtocol, WebSocketConnectionData, WebSocketServerConnectionProtocol, } from '@mswjs/interceptors/WebSocket' +/** + * @note A type-only import to prevent a runtime module cycle + * (the frame module imports this handler at runtime). + */ +import type { WebSocketNetworkFrameEventMap } from '../experimental/frames/websocket-frame' import { type Match, type Path, @@ -31,6 +37,14 @@ export interface WebSocketHandlerConnection { export interface WebSocketResolutionContext { baseUrl?: string + + /** + * An emit-only reference to the network frame's events. + * Allows handlers to emit additional events not covered by the frame + * into the network's life-cycle event stream (e.g. `server.events`). + */ + events?: Pick, 'emit'> + [kAutoConnect]?: boolean } @@ -217,8 +231,7 @@ export class WebSocketHandler { function createStopPropagationListener(handler: WebSocketHandler) { return function stopPropagationListener(event: Event) { const propagationStoppedAt = Reflect.get(event, 'kPropagationStoppedAt') as - | string - | undefined + string | undefined if (propagationStoppedAt && handler.id !== propagationStoppedAt) { event.stopImmediatePropagation() diff --git a/src/graphql/graphql-subscription.ts b/src/graphql/graphql-subscription.ts index 4c51cb372..2729e04b7 100644 --- a/src/graphql/graphql-subscription.ts +++ b/src/graphql/graphql-subscription.ts @@ -14,7 +14,9 @@ import { WebSocketHandler, kConnect, type WebSocketHandlerConnection, + type WebSocketResolutionContext, } from '#core/handlers/WebSocketHandler' +import { GraphQLSubscriptionEvent } from '#core/experimental/frames/websocket-frame' import { matchRequestUrl, type Path, @@ -196,6 +198,7 @@ interface GraphQLSubscriptionConnection { server: WebSocketServerConnectionProtocol subscribers: Map subscriptions: Map + events?: WebSocketResolutionContext['events'] isBound: boolean } @@ -239,6 +242,23 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { return this.#connections.get(clientId) } + public async run( + connection: WebSocketConnectionData, + resolutionContext?: WebSocketResolutionContext, + ): Promise { + const handlerConnection = await super.run(connection, resolutionContext) + + // Capture the network frame events reference for this connection. + // The transport emits life-cycle events (e.g. "graphql:subscription") + // long after the run: whenever the client sends a "subscribe" message. + if (handlerConnection) { + const transportConnection = this.#getOrCreateConnection(handlerConnection) + transportConnection.events = resolutionContext?.events + } + + return handlerConnection + } + /** * Send a `next` message with the given payload to the subscription. */ @@ -462,6 +482,7 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { for (const subscriber of connection.subscribers.values()) { if (subscriber({ node, message })) { + this.#emitSubscriptionEvent(connection, node, message) return } } @@ -472,6 +493,36 @@ export class GraphQLSubscriptionTransportHandler extends WebSocketHandler { toPublicUrl(connection.client.url), ) } + + /** + * Emit the "graphql:subscription" life-cycle event on the network. + * The event is emitted once the subscription has been established: + * matched by a subscription handler and resolved. + */ + #emitSubscriptionEvent( + connection: GraphQLSubscriptionConnection, + node: ParsedGraphQLQuery, + message: GraphQLWebSocketSubscribeMessage, + ): void { + // Anonymous subscriptions can never match a subscription handler. + if (!connection.events || !node.operationName) { + return + } + + connection.events.emit( + new GraphQLSubscriptionEvent({ + operationName: node.operationName, + query: message.payload.query, + variables: { ...message.payload.variables }, + request: new Request(connection.client.url, { + headers: { + connection: 'upgrade', + upgrade: 'websocket', + }, + }), + }), + ) + } } export type GraphQLSubscriptionName< diff --git a/test/node/graphql-api/events.test.ts b/test/node/graphql-api/events.test.ts new file mode 100644 index 000000000..91c07b36e --- /dev/null +++ b/test/node/graphql-api/events.test.ts @@ -0,0 +1,51 @@ +// @vitest-environment node +import { createClient } from 'graphql-ws' +import { setupServer } from 'msw/node' +import { graphql } from 'msw/graphql' + +const server = setupServer() + +beforeAll(() => { + server.listen() +}) + +afterEach(() => { + server.resetHandlers() + server.events.removeAllListeners() +}) + +afterAll(() => { + server.close() +}) + +it('emits the "graphql:subscription" event when a subscription is established', async () => { + const subscriptionListener = vi.fn() + server.events.on('graphql:subscription', subscriptionListener) + + const api = graphql.link('https://localhost/graphql') + server.use(api.subscription('OnCommentAdded', () => {})) + + const query = `subscription OnCommentAdded { commentAdded { text } }` + const client = createClient({ + url: 'wss://localhost/graphql', + }) + const subscription = client.iterate({ + query, + variables: { postId: 'post-1' }, + }) + + await expect.poll(() => subscriptionListener.mock.calls.length).toBe(1) + + expect(subscriptionListener).toHaveBeenCalledTimes(1) + + const [subscriptionEvent] = subscriptionListener.mock.calls[0] + expect(subscriptionEvent).toBeInstanceOf(Event) + expect(subscriptionEvent.type).toBe('graphql:subscription') + expect(subscriptionEvent.operationName).toBe('OnCommentAdded') + expect(subscriptionEvent.query).toBe(query) + expect(subscriptionEvent.variables).toEqual({ postId: 'post-1' }) + expect(subscriptionEvent.request).toBeInstanceOf(Request) + expect(subscriptionEvent.request.url).toBe('wss://localhost/graphql') + expect(subscriptionEvent.request.headers.get('connection')).toBe('upgrade') + expect(subscriptionEvent.request.headers.get('upgrade')).toBe('websocket') +}) diff --git a/test/node/graphql-api/graphql-subscription.test.ts b/test/node/graphql-api/graphql-subscription.test.ts index d3c2d9297..c7fda8005 100644 --- a/test/node/graphql-api/graphql-subscription.test.ts +++ b/test/node/graphql-api/graphql-subscription.test.ts @@ -26,6 +26,7 @@ beforeAll(() => { afterEach(() => { server.resetHandlers() + server.events.removeAllListeners() vi.restoreAllMocks() }) @@ -414,13 +415,18 @@ it('subscribes to extraneous pubsubs', async () => { const pubsub = createPubSub<{ commentAdded: [{ commentAdded: { text: string } }] }>() - const subscriptionEstablishedPromise = Promise.withResolvers() + const subscriptionAddedPromise = Promise.withResolvers() + + server.events.on('graphql:subscription', ({ operationName }) => { + if (operationName === 'OnCommentAdded') { + subscriptionAddedPromise.resolve() + } + }) const api = graphql.link('https://localhost/graphql') server.use( api.subscription('OnCommentAdded', ({ subscription }) => { subscription.from(pubsub.subscribe('commentAdded')) - subscriptionEstablishedPromise.resolve() }), api.mutation('AddComment', ({ variables }) => { const { comment } = variables @@ -452,7 +458,7 @@ it('subscribes to extraneous pubsubs', async () => { // pubsub. Unlike the client, the pubsub does not replay events published // before the subscription became active (the mutation below may otherwise // outrace the WebSocket handshake, e.g. on Node.js 24). - await subscriptionEstablishedPromise.promise + await subscriptionAddedPromise.promise const comment = { text: 'hello world' } await fetch('https://localhost/graphql', { @@ -482,6 +488,84 @@ it('subscribes to extraneous pubsubs', async () => { }) }) +it('emits the "graphql:subscription" life-cycle event when a subscription is established', async () => { + const subscriptionEventPromise = Promise.withResolvers<{ + operationName: string + query: string + variables: Record + request: Request + }>() + + server.events.on('graphql:subscription', (event) => { + subscriptionEventPromise.resolve(event) + }) + + const api = graphql.link('https://localhost/graphql') + server.use(api.subscription('OnCommentAdded', () => {})) + + const client = createClient({ + url: 'wss://localhost/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnCommentAdded($postId: ID!) { + commentAdded(postId: $postId) { + text + } + } + `, + variables: { postId: 'post-1' }, + }) + const pendingNext = subscription.next() + + const subscriptionEvent = await subscriptionEventPromise.promise + + expect(subscriptionEvent.operationName).toBe('OnCommentAdded') + expect(subscriptionEvent.query).toContain('subscription OnCommentAdded') + expect(subscriptionEvent.variables).toEqual({ postId: 'post-1' }) + expect(subscriptionEvent.request).toBeInstanceOf(Request) + expect(subscriptionEvent.request.url).toBe('wss://localhost/graphql') + expect(subscriptionEvent.request.headers.get('connection')).toBe('upgrade') + expect(subscriptionEvent.request.headers.get('upgrade')).toBe('websocket') + + pendingNext.catch(() => {}) + await client.dispose() +}) + +it('does not emit the "graphql:subscription" life-cycle event for unhandled subscriptions', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + + const subscriptionListener = vi.fn() + server.events.on('graphql:subscription', subscriptionListener) + + const api = graphql.link('https://localhost/graphql') + server.use(api.subscription('OnCommentAdded', () => {})) + + const client = createClient({ + url: 'wss://localhost/graphql', + }) + const subscription = client.iterate({ + query: gql` + subscription OnPostAdded { + postAdded { + id + } + } + `, + }) + const pendingNext = subscription.next() + + // The unhandled subscription warning marks the dispatch completion. + await expect + .poll(() => vi.mocked(console.warn).mock.calls.flat().join('\n')) + .toMatch(/no matching subscription handler/) + + expect(subscriptionListener).not.toHaveBeenCalled() + + pendingNext.catch(() => {}) + await client.dispose() +}) + it('combines extraneous and default pubsubs', async () => { const pubsub = createPubSub<{ commentAdded: [{ commentAdded: { text: string } }] From 1324bd5ff8f41c3ba0c663ea7d04c5ec4fb173a7 Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 21:31:18 +0200 Subject: [PATCH 6/8] test(graphql): add failing `graphql.operation` test --- .../graphql-api/graphql-subscription.test.ts | 34 ++++++++++++++++--- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/test/node/graphql-api/graphql-subscription.test.ts b/test/node/graphql-api/graphql-subscription.test.ts index c7fda8005..776baf0c7 100644 --- a/test/node/graphql-api/graphql-subscription.test.ts +++ b/test/node/graphql-api/graphql-subscription.test.ts @@ -206,9 +206,6 @@ it('scopes published data to the subscribed client', async () => { return messages } - // Each client must receive only the data published to its - // own subscription (subscription ids are unique per socket, - // not across the clients). await expect(collectMessages('john')).resolves.toEqual([ { data: { greeting: 'hello, john' } }, ]) @@ -245,7 +242,6 @@ it('respects handler overrides for the same operation', async () => { `, }) - // Must receive the data from the override handler. await expect(subscription.next()).resolves.toEqual({ done: false, value: { @@ -255,10 +251,38 @@ it('respects handler overrides for the same operation', async () => { }, }) - // The initial handler must not be called (first matching handler wins). expect(initialResolver).not.toHaveBeenCalled() }) +it('matches an outgoing subscription with "graphql.operation()"', async () => { + const operationResolver = vi.fn() + + const api = graphql.link('https://localhost/graphql') + server.use(api.operation(operationResolver)) + + const client = createClient({ + url: 'wss://localhost/graphql', + }) + client.iterate({ + query: gql` + subscription OnCommentAdded($postId: ID!) { + commentAdded(postId: $postId) { + text + } + } + `, + variables: { postId: 'post-1' }, + }) + + await expect + .poll(() => operationResolver) + .toHaveBeenCalledExactlyOnceWith({ + operationName: 'OnCommentAdded', + query: expect.stringContaining('subscription OnCommentAdded'), + variables: { postId: 'post-1' }, + }) +}) + it('supports one-time subscription handlers', async () => { vi.spyOn(console, 'warn').mockImplementation(() => {}) From e39a3ec341fd9d9121965d35e2c6b5347d325a17 Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 21:45:19 +0200 Subject: [PATCH 7/8] fix: skip sibling handlers from `.listHandlers()` --- src/core/experimental/define-network.ts | 2 +- .../experimental/handlers-controller.test.ts | 29 +++++++++++++++++++ src/core/experimental/handlers-controller.ts | 17 ++++++++++- src/core/experimental/setup-api.ts | 2 +- .../utils/internal/attachSiblingHandlers.ts | 20 +++++++++++++ 5 files changed, 67 insertions(+), 3 deletions(-) diff --git a/src/core/experimental/define-network.ts b/src/core/experimental/define-network.ts index f792f505e..089bed6ea 100644 --- a/src/core/experimental/define-network.ts +++ b/src/core/experimental/define-network.ts @@ -238,7 +238,7 @@ export function defineNetwork>>( handlersController.restore() }, listHandlers() { - return toReadonlyArray(handlersController.currentHandlers()) + return toReadonlyArray(handlersController.listHandlers()) }, } } diff --git a/src/core/experimental/handlers-controller.test.ts b/src/core/experimental/handlers-controller.test.ts index edd33bccd..0ac402595 100644 --- a/src/core/experimental/handlers-controller.test.ts +++ b/src/core/experimental/handlers-controller.test.ts @@ -308,6 +308,35 @@ describe(InMemoryHandlersController.prototype.reset, () => { }) }) +describe(InMemoryHandlersController.prototype.listHandlers, () => { + it('lists explicitly registered handlers, hiding their siblings', () => { + const httpHandler = http.get('/', () => {}) + const wsHandler = ws.link('*').addEventListener('connection', () => {}) + + const controller = new InMemoryHandlersController([httpHandler, wsHandler]) + + expect(controller.listHandlers()).toEqual([httpHandler, wsHandler]) + }) + + it('hides siblings of runtime handlers', () => { + const controller = new InMemoryHandlersController([]) + const wsHandler = ws.link('*').addEventListener('connection', () => {}) + + controller.use([wsHandler]) + + expect(controller.listHandlers()).toEqual([wsHandler]) + }) + + it('keeps siblings in "currentHandlers()" so the handler lifecycle reaches them', () => { + const wsHandler = ws.link('*').addEventListener('connection', () => {}) + const [upgradeHandler] = getSiblingHandlers(wsHandler) + + const controller = new InMemoryHandlersController([wsHandler]) + + expect(controller.currentHandlers()).toEqual([wsHandler, upgradeHandler]) + }) +}) + describe(InMemoryHandlersController.prototype.getHandlersByKind, () => { it('returns an empty array given an empty controller', () => { const controller = new InMemoryHandlersController([]) diff --git a/src/core/experimental/handlers-controller.ts b/src/core/experimental/handlers-controller.ts index d400eba60..f3be64084 100644 --- a/src/core/experimental/handlers-controller.ts +++ b/src/core/experimental/handlers-controller.ts @@ -2,7 +2,10 @@ import { invariant } from 'outvariant' import { type RequestHandler } from '../handlers/RequestHandler' import { type WebSocketHandler } from '../handlers/WebSocketHandler' import { devUtils } from '../utils/internal/devUtils' -import { getSiblingHandlers } from '../utils/internal/attachSiblingHandlers' +import { + getSiblingHandlers, + isSiblingHandler, +} from '../utils/internal/attachSiblingHandlers' export type AnyHandler = RequestHandler | WebSocketHandler export type HandlersMap = Partial>> @@ -70,6 +73,18 @@ export abstract class HandlersController { .filter((handler) => handler != null) } + /** + * Return the list of explicitly registered handlers. + * Unlike `currentHandlers()`, this excludes sibling handlers, + * which are an implementation detail (e.g. the WebSocket upgrade + * handler or the GraphQL subscription transport). + */ + public listHandlers(): Array { + return this.currentHandlers().filter((handler) => { + return !isSiblingHandler(handler) + }) + } + public getHandlersByKind(kind: AnyHandler['kind']): Array { return this.getState().handlers[kind] || [] } diff --git a/src/core/experimental/setup-api.ts b/src/core/experimental/setup-api.ts index ece7e44ef..e63fe0c80 100644 --- a/src/core/experimental/setup-api.ts +++ b/src/core/experimental/setup-api.ts @@ -50,6 +50,6 @@ export abstract class SetupApi< } public listHandlers(): ReadonlyArray { - return toReadonlyArray(this.handlersController.currentHandlers()) + return toReadonlyArray(this.handlersController.listHandlers()) } } diff --git a/src/core/utils/internal/attachSiblingHandlers.ts b/src/core/utils/internal/attachSiblingHandlers.ts index e7a63e374..4e94cf07f 100644 --- a/src/core/utils/internal/attachSiblingHandlers.ts +++ b/src/core/utils/internal/attachSiblingHandlers.ts @@ -2,6 +2,7 @@ import { invariant } from 'outvariant' import type { AnyHandler } from '../../experimental/handlers-controller' const kSiblingHandlers = Symbol('kSiblingHandlers') +const kIsSiblingHandler = Symbol('kIsSiblingHandler') export function attachSiblingHandlers( owner: T, @@ -20,9 +21,28 @@ export function attachSiblingHandlers( configurable: false, }) + // Mark the siblings so introspection (e.g. `.listHandlers()`) can tell + // explicitly registered handlers from their implementation details. + // Shared siblings can be attached to multiple owners, so skip the + // already-marked ones. + for (const sibling of siblings) { + if (!isSiblingHandler(sibling)) { + Object.defineProperty(sibling, kIsSiblingHandler, { + value: true, + enumerable: false, + writable: false, + configurable: false, + }) + } + } + return owner } export function getSiblingHandlers(owner: AnyHandler): Array { return Reflect.get(owner, kSiblingHandlers) || [] } + +export function isSiblingHandler(handler: AnyHandler): boolean { + return Reflect.get(handler, kIsSiblingHandler) === true +} From caaba29d97ea453fc0e0dd9c797553806eb72f9a Mon Sep 17 00:00:00 2001 From: Artem Zakharchenko Date: Fri, 10 Jul 2026 21:45:45 +0200 Subject: [PATCH 8/8] fix: make `graphql.operation` catch subscriptions --- src/graphql/graphql.ts | 56 +++++++++++++++++++++++++++++++++++++++--- 1 file changed, 52 insertions(+), 4 deletions(-) diff --git a/src/graphql/graphql.ts b/src/graphql/graphql.ts index 6ac45fd16..5497f474d 100644 --- a/src/graphql/graphql.ts +++ b/src/graphql/graphql.ts @@ -13,6 +13,7 @@ import { type GraphQLPredicate, } from './graphql-handler' import type { Path } from '#core/utils/matching/matchRequestUrl' +import { attachSiblingHandlers } from '#core/utils/internal/attachSiblingHandlers' import { createGraphQLSubscriptionHandler, type GraphQLSubscriptionHandlerFactory, @@ -59,9 +60,44 @@ function createScopedGraphQLHandler( } } -function createGraphQLOperationHandler(url: Path): GraphQLOperationHandler { +function createGraphQLOperationHandler( + url: Path, + subscriptionFactory?: GraphQLSubscriptionHandlerFactory, +): GraphQLOperationHandler { return (resolver, options) => { - return new GraphQLHandler('all', new RegExp('.*'), url, resolver, options) + const handler = new GraphQLHandler( + 'all', + new RegExp('.*'), + url, + resolver, + options, + ) + + if (!subscriptionFactory) { + return handler + } + + // Attach a catch-all subscription handler as a sibling so the + // operation handler also matches GraphQL subscriptions over WebSocket. + // The subscription transport guarantees only actual GraphQL + // subscriptions are dispatched to it. + const subscriptionCatchAllHandler = subscriptionFactory( + new RegExp('.*'), + ({ operationName, subscription }) => { + /** + * @note Subscriptions are resolved imperatively so the resolver + * is invoked with a reduced info object (no request to expose), + * and its return value is ignored. + */ + resolver({ + operationName, + query: subscription.query, + variables: subscription.variables, + } as Parameters[0]) + }, + ) + + return attachSiblingHandlers(handler, [subscriptionCatchAllHandler]) } } @@ -125,6 +161,11 @@ export const graphql = { * return HttpResponse.json({ data: { name: 'John' } }) * }) * + * @note Unlike `graphql.link(url).operation()`, this handler does not + * match GraphQL subscriptions: intercepting them requires claiming the + * WebSocket connections to a concrete endpoint, and a wildcard would + * claim every WebSocket connection on the page. + * * @see {@link https://mswjs.io/docs/api/graphql#graphqloperationresolver `graphql.operation()` API reference} */ operation: createGraphQLOperationHandler('*'), @@ -139,14 +180,21 @@ export const graphql = { * @see {@link https://mswjs.io/docs/api/graphql#graphqllinkurl `graphql.link()` API reference} */ link(url: Path): GraphQLLinkHandlers { + /** + * @note Create the subscription handler factory once per link so + * the `subscription()` and `operation()` handlers share the same + * underlying subscription transport (deduped by reference). + */ + const subscription = createGraphQLSubscriptionHandler(url) + return { - operation: createGraphQLOperationHandler(url), + operation: createGraphQLOperationHandler(url, subscription), query: createScopedGraphQLHandler('query' as OperationTypeNode, url), mutation: createScopedGraphQLHandler( 'mutation' as OperationTypeNode, url, ), - subscription: createGraphQLSubscriptionHandler(url), + subscription, } }, }