diff --git a/packages/javascript-api/src/lib/services/graphql/__tests__/subscription-protocol.spec.ts b/packages/javascript-api/src/lib/services/graphql/__tests__/subscription-protocol.spec.ts new file mode 100644 index 00000000..01005f20 --- /dev/null +++ b/packages/javascript-api/src/lib/services/graphql/__tests__/subscription-protocol.spec.ts @@ -0,0 +1,107 @@ +import { LegacySubscriptionProtocol } from '../subscription-protocol'; + +describe('LegacySubscriptionProtocol', () => { + const protocol = new LegacySubscriptionProtocol(); + + describe('outgoing frames', () => { + it('serializes connection_init', () => { + expect(JSON.parse(protocol.serializeConnectionInit())).toEqual({ + type: 'connection_init', + payload: null, + }); + }); + + it('serializes subscribe as start', () => { + expect( + JSON.parse(protocol.serializeSubscribe('7', 'subscription { baba }')), + ).toEqual({ + id: '7', + type: 'start', + payload: { query: 'subscription { baba }' }, + }); + }); + + it('serializes unsubscribe as stop', () => { + expect(JSON.parse(protocol.serializeUnsubscribe('7'))).toEqual({ + id: '7', + type: 'stop', + payload: null, + }); + }); + + it('serializes ping without id and payload', () => { + expect(JSON.parse(protocol.serializePing())).toEqual({ type: 'ping' }); + }); + }); + + describe('incoming frames', () => { + it('parses ka as keep-alive', () => { + expect(protocol.parseIncomingMessage('{"type":"ka"}')).toEqual({ + type: 'keep-alive', + }); + }); + + it('parses connection_ack', () => { + expect( + protocol.parseIncomingMessage('{"type":"connection_ack"}'), + ).toEqual({ type: 'connection-ack' }); + }); + + it('parses pong', () => { + expect(protocol.parseIncomingMessage('{"type":"pong"}')).toEqual({ + type: 'pong', + }); + }); + + it('parses data', () => { + expect( + protocol.parseIncomingMessage( + '{"id":"1","type":"data","payload":{"data":{"baba":9}}}', + ), + ).toEqual({ type: 'data', id: '1', data: { baba: 9 } }); + }); + + it('parses data without payload', () => { + expect(protocol.parseIncomingMessage('{"id":"1","type":"data"}')).toEqual( + { type: 'data', id: '1', data: undefined }, + ); + }); + + it('parses complete', () => { + expect( + protocol.parseIncomingMessage('{"id":"1","type":"complete"}'), + ).toEqual({ type: 'complete', id: '1' }); + }); + + it('parses error with errors payload', () => { + expect( + protocol.parseIncomingMessage( + '{"id":"1","type":"error","payload":{"errors":[{"message":"boom"}]}}', + ), + ).toEqual({ + type: 'subscription-error', + id: '1', + errors: [{ message: 'boom' }], + }); + }); + + it('parses error without errors payload as empty error list', () => { + expect( + protocol.parseIncomingMessage('{"id":"1","type":"error"}'), + ).toEqual({ type: 'subscription-error', id: '1', errors: [] }); + }); + + it('parses an unknown message type as unrecognized, keeping payload', () => { + expect( + protocol.parseIncomingMessage( + '{"id":"1","type":"surprise","payload":{"data":{"a":1},"errors":[{"message":"boom"}]}}', + ), + ).toEqual({ + type: 'unrecognized', + id: '1', + data: { a: 1 }, + errors: [{ message: 'boom' }], + }); + }); + }); +}); diff --git a/packages/javascript-api/src/lib/services/graphql/graphql.service.ts b/packages/javascript-api/src/lib/services/graphql/graphql.service.ts index 6c3127c0..b38c4980 100644 --- a/packages/javascript-api/src/lib/services/graphql/graphql.service.ts +++ b/packages/javascript-api/src/lib/services/graphql/graphql.service.ts @@ -1,9 +1,4 @@ -import { - DocumentNode, - GraphQLErrorExtensions, - print, - SourceLocation, -} from 'graphql'; +import { DocumentNode, print } from 'graphql'; import WebSocket, { CloseEvent } from 'isomorphic-ws'; import { BehaviorSubject, @@ -24,6 +19,13 @@ import { calculateRandomizedExponentialBackoffTime } from '../../util/randomized import { sleepMs } from '../../util/sleep-ms/sleep-ms.js'; import { ApiBase, GraphqlQuery } from '../api-base/api-base.js'; import { TemporaryApiKeyService } from '../temporary-api-key/temporary-api-key.service.js'; +import { + LegacySubscriptionProtocol, + QminderGraphQLError, + SubscriptionProtocol, +} from './subscription-protocol.js'; + +export type { QminderGraphQLError } from './subscription-protocol.js'; function parseQuery(queryOrDocumentNode: string | DocumentNode): string { if (typeof queryOrDocumentNode === 'string') { @@ -37,46 +39,11 @@ function parseQuery(queryOrDocumentNode: string | DocumentNode): string { throw new Error('queryOrDocumentNode must be a string or a DocumentNode'); } -export interface QminderGraphQLError { - readonly message: string; - readonly errorType?: string | null; - readonly extensions?: GraphQLErrorExtensions | null; - readonly sourcePreview?: string | null; - readonly offendingToken?: string | null; - readonly locations?: SourceLocation[] | null; - readonly path?: (string | number)[] | null; -} - -interface Message { - readonly id?: string; - readonly type: MessageType; - readonly payload?: { - readonly data?: Record | null; - readonly errors?: QminderGraphQLError[]; - }; -} - interface Subscription { readonly messageId: string; readonly query: string; } -enum MessageType { - // To Server - GQL_CONNECTION_INIT = 'connection_init', - GQL_START = 'start', - GQL_STOP = 'stop', - GQL_PING = 'ping', - - // From Server - GQL_CONNECTION_ACK = 'connection_ack', - GQL_DATA = 'data', - GQL_CONNECTION_KEEP_ALIVE = 'ka', - GQL_COMPLETE = 'complete', - GQL_PONG = 'pong', - GQL_ERROR = 'error', -} - const NON_RETRYABLE_SUBSCRIPTION_ERROR_TYPES = [ 'BAD_REQUEST', 'FIELD_NOT_FOUND', @@ -113,6 +80,8 @@ const BLOCKED_RETRY_INTERVAL_MS = 5 * 60 * 1000; */ export class GraphqlService { private readonly logger = new Logger('GraphQL'); + private readonly protocol: SubscriptionProtocol = + new LegacySubscriptionProtocol(); private apiServer: string; private socket: WebSocket = null; @@ -312,17 +281,17 @@ export class GraphqlService { const messageId = `${++this.subscriptionsCount}`; this.subscriptions.push({ messageId, query }); - this.sendMessage(messageId, MessageType.GQL_START, { query }).catch( - (error: Error) => { - this.logger.error('Failed to start subscription: ', error); - }, - ); + this.sendMessage( + this.protocol.serializeSubscribe(messageId, query), + ).catch((error: Error) => { + this.logger.error('Failed to start subscription: ', error); + }); this.messagesSubscribers.set(messageId, subscriber); return () => { if (this.messagesSubscribers.has(messageId)) { - this.sendMessage(messageId, MessageType.GQL_STOP, null).catch( + this.sendMessage(this.protocol.serializeUnsubscribe(messageId)).catch( (error) => { this.logger.error('Failed to stop subscription: ', error); }, @@ -434,13 +403,7 @@ export class GraphqlService { this.socket = new WebSocket(this.getServerUrl(temporaryApiKey)); this.socket.onopen = () => { - this.sendRawMessage( - JSON.stringify({ - id: undefined, - type: MessageType.GQL_CONNECTION_INIT, - payload: null, - }), - ); + this.sendRawMessage(this.protocol.serializeConnectionInit()); }; this.socket.onclose = (event) => { @@ -512,13 +475,13 @@ export class GraphqlService { return; } - const message: Message = JSON.parse(event.data); + const message = this.protocol.parseIncomingMessage(event.data); switch (message.type) { - case MessageType.GQL_CONNECTION_KEEP_ALIVE: + case 'keep-alive': break; - case MessageType.GQL_CONNECTION_ACK: { + case 'connection-ack': { this.clearErroredSubscriptionsTimeouts(); this.retryableErroredSubscriptionsRetryCount = 0; this.retryableErroredSubscriptionsAction$.next({ type: 'clear' }); @@ -533,11 +496,7 @@ export class GraphqlService { let resubscriptionFailed = false; for (const { messageId, query } of this.subscriptions) { - const msg = JSON.stringify({ - id: messageId, - type: MessageType.GQL_START, - payload: { query }, - }); + const msg = this.protocol.serializeSubscribe(messageId, query); if (!this.sendRawMessage(msg)) { this.logger.warn( @@ -561,16 +520,16 @@ export class GraphqlService { break; } - case MessageType.GQL_DATA: + case 'data': this.retryableErroredSubscriptionsAction$.next({ type: 'remove', messageId: message.id, }); - this.messagesSubscribers.get(message.id)?.next(message.payload?.data); + this.messagesSubscribers.get(message.id)?.next(message.data); break; - case MessageType.GQL_COMPLETE: { + case 'complete': { this.retryableErroredSubscriptionsAction$.next({ type: 'remove', messageId: message.id, @@ -583,13 +542,13 @@ export class GraphqlService { break; } - case MessageType.GQL_PONG: + case 'pong': clearTimeout(this.pongTimeout); this.connectionAttemptsCount = 0; break; - case MessageType.GQL_ERROR: { - const errors = message.payload?.errors ?? []; + case 'subscription-error': { + const errors = message.errors; if (this.isAnySubscriptionErrorNonRetryable(errors)) { this.logger.error( @@ -636,19 +595,24 @@ export class GraphqlService { break; } - default: { + case 'unrecognized': { const subscriber = this.messagesSubscribers.get(message.id); if (!subscriber) { return; } - if (message.payload?.data) { + if (message.data) { this.cleanUpSubscription(message.id); - subscriber.error(message.payload.data); - } else if (message.payload?.errors?.length) { + subscriber.error(message.data); + } else if (message.errors?.length) { this.cleanUpSubscription(message.id); - subscriber.error(message.payload.errors); + subscriber.error(message.errors); } + break; + } + default: { + const exhaustiveCheck: never = message; + return exhaustiveCheck; } } }; @@ -660,17 +624,13 @@ export class GraphqlService { ); } - private async sendMessage( - id: string, - type: MessageType, - payload: Record | null, - ): Promise { + private async sendMessage(message: string): Promise { if (this.connectionStatus !== ConnectionStatus.CONNECTED) { await this.openSocket(); return; } - if (!this.sendRawMessage(JSON.stringify({ id, type, payload }))) { + if (!this.sendRawMessage(message)) { this.logger.warn('Message dropped: WebSocket is not in OPEN state'); await this.handleConnectionDrop(); } @@ -751,7 +711,7 @@ export class GraphqlService { }); }, PONG_TIMEOUT_IN_MS); - this.sendRawMessage(JSON.stringify({ type: MessageType.GQL_PING })); + this.sendRawMessage(this.protocol.serializePing()); } private async handleConnectionDrop(): Promise { @@ -874,11 +834,10 @@ export class GraphqlService { } this.sendRawMessage( - JSON.stringify({ - id: subscription.messageId, - type: MessageType.GQL_START, - payload: { query: subscription.query }, - }), + this.protocol.serializeSubscribe( + subscription.messageId, + subscription.query, + ), ); } diff --git a/packages/javascript-api/src/lib/services/graphql/subscription-protocol.ts b/packages/javascript-api/src/lib/services/graphql/subscription-protocol.ts new file mode 100644 index 00000000..9f9c8287 --- /dev/null +++ b/packages/javascript-api/src/lib/services/graphql/subscription-protocol.ts @@ -0,0 +1,144 @@ +import { GraphQLErrorExtensions, SourceLocation } from 'graphql'; + +export interface QminderGraphQLError { + readonly message: string; + readonly errorType?: string | null; + readonly extensions?: GraphQLErrorExtensions | null; + readonly sourcePreview?: string | null; + readonly offendingToken?: string | null; + readonly locations?: SourceLocation[] | null; + readonly path?: (string | number)[] | null; +} + +/** + * A message received from the subscription server, normalized so that + * GraphqlService does not depend on the wire format of a concrete protocol. + */ +export type IncomingSubscriptionMessage = + | { readonly type: 'connection-ack' } + | { readonly type: 'keep-alive' } + | { readonly type: 'pong' } + | { + readonly type: 'data'; + readonly id?: string; + readonly data?: Record | null; + } + | { readonly type: 'complete'; readonly id?: string } + | { + readonly type: 'subscription-error'; + readonly id?: string; + readonly errors: QminderGraphQLError[]; + } + | { + readonly type: 'unrecognized'; + readonly id?: string; + readonly data?: Record | null; + readonly errors?: QminderGraphQLError[]; + }; + +/** + * The wire protocol spoken over the subscription WebSocket: serializes + * outgoing frames and normalizes incoming ones. + */ +export interface SubscriptionProtocol { + serializeConnectionInit(): string; + serializeSubscribe(id: string, query: string): string; + serializeUnsubscribe(id: string): string; + serializePing(): string; + parseIncomingMessage(data: string): IncomingSubscriptionMessage; +} + +enum LegacyMessageType { + // To Server + GQL_CONNECTION_INIT = 'connection_init', + GQL_START = 'start', + GQL_STOP = 'stop', + GQL_PING = 'ping', + + // From Server + GQL_CONNECTION_ACK = 'connection_ack', + GQL_DATA = 'data', + GQL_CONNECTION_KEEP_ALIVE = 'ka', + GQL_COMPLETE = 'complete', + GQL_PONG = 'pong', + GQL_ERROR = 'error', +} + +interface LegacyMessage { + readonly id?: string; + readonly type: LegacyMessageType; + readonly payload?: { + readonly data?: Record | null; + readonly errors?: QminderGraphQLError[]; + }; +} + +/** + * The protocol of the original Qminder subscription endpoint, a dialect of the + * deprecated `subscriptions-transport-ws` protocol (with graphql-ws style + * ping/pong on top). + */ +export class LegacySubscriptionProtocol implements SubscriptionProtocol { + serializeConnectionInit(): string { + return JSON.stringify({ + type: LegacyMessageType.GQL_CONNECTION_INIT, + payload: null, + }); + } + + serializeSubscribe(id: string, query: string): string { + return JSON.stringify({ + id, + type: LegacyMessageType.GQL_START, + payload: { query }, + }); + } + + serializeUnsubscribe(id: string): string { + return JSON.stringify({ + id, + type: LegacyMessageType.GQL_STOP, + payload: null, + }); + } + + serializePing(): string { + return JSON.stringify({ type: LegacyMessageType.GQL_PING }); + } + + parseIncomingMessage(data: string): IncomingSubscriptionMessage { + const message: LegacyMessage = JSON.parse(data); + + switch (message.type) { + case LegacyMessageType.GQL_CONNECTION_KEEP_ALIVE: + return { type: 'keep-alive' }; + + case LegacyMessageType.GQL_CONNECTION_ACK: + return { type: 'connection-ack' }; + + case LegacyMessageType.GQL_PONG: + return { type: 'pong' }; + + case LegacyMessageType.GQL_DATA: + return { type: 'data', id: message.id, data: message.payload?.data }; + + case LegacyMessageType.GQL_COMPLETE: + return { type: 'complete', id: message.id }; + + case LegacyMessageType.GQL_ERROR: + return { + type: 'subscription-error', + id: message.id, + errors: message.payload?.errors ?? [], + }; + + default: + return { + type: 'unrecognized', + id: message.id, + data: message.payload?.data, + errors: message.payload?.errors, + }; + } + } +}