GraphQL: Subscriptions, Security & Plugins

npm i graphql-ws graphql-subscriptions

GraphQLModule.forRoot<ApolloDriverConfig>({
  driver: ApolloDriver,
  autoSchemaFile: true,
  subscriptions: { 'graphql-ws': true },
}),

import { Global, Module } from '@nestjs/common';
import { PubSub } from 'graphql-subscriptions';

export const PUB_SUB = 'PUB_SUB';

@Global()
@Module({
  providers: [{ provide: PUB_SUB, useValue: new PubSub() }],
  exports: [PUB_SUB],
})
export class PubSubModule {}

import { moderatorsOnly } from './field-middleware.js';
import { Field, ID, InputType, Int, ObjectType } from '@nestjs/graphql';

@ObjectType()
export class Review {
  @Field(() => ID)
  id!: string;

  @Field(() => Int)
  bookId!: number;

  @Field({ nullable: true, middleware: [moderatorsOnly] })
  author?: string;

  @Field({ complexity: 3 })
  text!: string;
}

@InputType()
export class NewReviewInput {
  @Field(() => Int)
  bookId!: number;

  @Field()
  text!: string;
}

import { BadRequestException, Inject } from '@nestjs/common';
import { Args, Int, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
import { PubSub } from 'graphql-subscriptions';
import { CurrentUser } from './current-user.decorator.js';
import type { User } from './users.service.js';
import { NewReviewInput, Review } from './review.model.js';
import { ReviewTooLongError } from './domain-error.filter.js';
import { PUB_SUB } from './pubsub.module.js';

export const REVIEW_ADDED = 'reviewAdded';

@Resolver(() => Review)
export class ReviewsResolver {
  private readonly reviews: Review[] = [];

  constructor(@Inject(PUB_SUB) private readonly pubSub: PubSub) {}

  @Query(() => [Review])
  reviewsFor(@Args('bookId', { type: () => Int }) bookId: number) {
    return this.reviews.filter((review) => review.bookId === bookId);
  }

  @Mutation(() => Review)
  async addReview(@Args('input') input: NewReviewInput, @CurrentUser() user: User) {
    if (input.text.length > 200) {
      throw new ReviewTooLongError(200);
    }
    const review: Review = {
      id: String(this.reviews.length + 1),
      bookId: input.bookId,
      author: user.name,
      text: input.text,
    };
    this.reviews.push(review);
    // The payload key must match the subscription field name (no `resolve` here).
    await this.pubSub.publish(REVIEW_ADDED, { reviewAdded: review });
    return review;
  }

  @Subscription(() => Review, {
    filter(this: ReviewsResolver, payload: { reviewAdded: Review }, variables: { bookId: number }) {
      return payload.reviewAdded.bookId === variables.bookId;
    },
  })
  reviewAdded(@Args('bookId', { type: () => Int }) _bookId: number) {
    return this.pubSub.asyncIterableIterator(REVIEW_ADDED);
  }

  // Same event, but `resolve` unwraps the payload and can shape the value.
  @Subscription(() => String, {
    name: 'reviewTextAdded',
    resolve: (payload: Review) => payload.text.toUpperCase(),
  })
  reviewTextAdded() {
    return this.pubSub.asyncIterableIterator('reviewTextRaw');
  }

  @Query(() => String)
  explode(@Args('kind') kind: string): string {
    if (kind === 'http') {
      throw new BadRequestException('Unsupported kind');
    }
    throw new Error('connect ECONNREFUSED 10.0.0.5:5432');
  }

  @Mutation(() => Boolean)
  async shout(@Args('text') text: string) {
    await this.pubSub.publish('reviewTextRaw', { text });
    return true;
  }
}

import { createClient } from 'graphql-ws';
import WebSocket from 'ws';

const client = createClient({
  url: `ws://127.0.0.1:${port}/graphql`,
  webSocketImpl: WebSocket,
  retryAttempts: 0,
  // Sent once, in the connection_init message: this is what onConnect receives.
  connectionParams: { authorization: 'Bearer <session id>' },
});

client.subscribe(
  {
    query: 'subscription ($bookId: Int!) { reviewAdded(bookId: $bookId) { id bookId text author } }',
    variables: { bookId: 1 },
  },
  { next: (result) => console.log(result), error: console.error, complete: () => undefined },
);

{ "data": { "reviewAdded": { "id": "2", "bookId": 1, "text": "great", "author": null } } }

import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo';
import { Module } from '@nestjs/common';
import { APP_FILTER, APP_GUARD, APP_INTERCEPTOR } from '@nestjs/core';
import { GraphQLModule } from '@nestjs/graphql';
import { GraphQLFormattedError } from 'graphql';
import type { Context } from 'graphql-ws';
import { ComplexityPlugin, RequestStats, RequestStatsPlugin } from './apollo-plugins.js';
import { ReviewTooLongFilter } from './domain-error.filter.js';
import { recordFields } from './field-middleware.js';
import { GqlAuthGuard } from './gql-auth.guard.js';
import { PubSubModule } from './pubsub.module.js';
import { ReviewsResolver } from './reviews.resolver.js';
import { TimingInterceptor } from './timing.interceptor.js';
import { UsersModule } from './users.module.js';
import { UsersService } from './users.service.js';

@Module({
  imports: [
    PubSubModule,
    UsersModule,
    GraphQLModule.forRootAsync<ApolloDriverConfig>({
      driver: ApolloDriver,
      imports: [UsersModule],
      inject: [UsersService],
      useFactory: (users: UsersService) => ({
        autoSchemaFile: true,
        subscriptions: {
          'graphql-ws': {
            // Runs once per WebSocket connection, before any subscribe message.
            onConnect: (ctx: Context) => {
              const { connectionParams, extra } = ctx as Context & { extra: { user?: unknown } };
              const user = users.fromAuthorization(connectionParams?.authorization as string | undefined);
              if (!user) {
                return false; // the server closes the socket with code 4403
              }
              extra.user = user;
            },
          },
        },
        // HTTP requests provide `req`; graphql-ws provides `extra` (set in onConnect).
        context: ({ req, extra }: { req?: { headers: Record<string, string> }; extra?: { user?: unknown } }) => ({
          req,
          user: extra?.user ?? users.fromAuthorization(req?.headers.authorization),
        }),
        buildSchemaOptions: { fieldMiddleware: [recordFields] },
        includeStacktraceInErrorResponses: false,
        formatError: (formatted: GraphQLFormattedError): GraphQLFormattedError => {
          const { code } = (formatted.extensions ?? {}) as { code?: string };
          if (code === 'INTERNAL_SERVER_ERROR') {
            return { message: 'Something went wrong', extensions: { code } };
          }
          return formatted;
        },
      }),
    }),
  ],
  providers: [
    ReviewsResolver,
    RequestStats,
    RequestStatsPlugin,
    ComplexityPlugin,
    { provide: APP_GUARD, useClass: GqlAuthGuard },
    { provide: APP_INTERCEPTOR, useClass: TimingInterceptor },
    { provide: APP_FILTER, useClass: ReviewTooLongFilter },
  ],
})
export class SubscriptionsAppModule {}

closed { code: 4403, reason: 'Forbidden' }

npm i graphql-redis-subscriptions ioredis

import { ConfigService } from '@nestjs/config';
import { RedisPubSub } from 'graphql-redis-subscriptions';
import { PUB_SUB } from './pubsub.module.js';

/** Drop-in replacement for the in-memory provider: same token, same API. */
export const redisPubSubProvider = {
  provide: PUB_SUB,
  inject: [ConfigService],
  useFactory: (config: ConfigService) =>
    // The library creates two connections itself: one to publish and one to
    // subscribe (a subscribed connection cannot run other commands).
    new RedisPubSub({ connection: config.getOrThrow<string>('REDIS_URL') }),
};

import { Module } from '@nestjs/common';
import { Args, Context, GraphQLModule, Int, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
import { MercuriusDriver, MercuriusDriverConfig } from '@nestjs/mercurius';
import type { PubSub } from 'mercurius';
import { Review } from './review.model.js';

export const hookLog: string[] = [];

@Resolver(() => Review)
class MercuriusReviewsResolver {
  @Query(() => String)
  status() {
    return 'ok';
  }

  @Mutation(() => Review)
  async addReview(
    @Args('bookId', { type: () => Int }) bookId: number,
    @Args('text') text: string,
    @Context('pubsub') pubSub: PubSub,
  ) {
    const review = { id: '1', bookId, text } as Review;
    await pubSub.publish({ topic: 'reviewAdded', payload: { reviewAdded: review } });
    return review;
  }

  @Subscription(() => Review, {
    filter: (payload: { reviewAdded: Review }, variables: { bookId: number }) =>
      payload.reviewAdded.bookId === variables.bookId,
  })
  reviewAdded(
    @Args('bookId', { type: () => Int }) _bookId: number,
    @Context('pubsub') pubSub: PubSub,
  ) {
    return pubSub.subscribe('reviewAdded');
  }
}

@Module({
  imports: [
    GraphQLModule.forRoot<MercuriusDriverConfig>({
      driver: MercuriusDriver,
      autoSchemaFile: true,
      subscription: true,
      hooks: {
        preExecution: async (_schema, document) => {
          const op = document.definitions[0];
          hookLog.push(`preExecution ${op.kind === 'OperationDefinition' ? op.operation : op.kind}`);
        },
        onResolution: async (execution) => {
          hookLog.push(`onResolution ${execution.errors ? 'errors' : 'data'}`);
        },
        preSubscriptionExecution: async () => {
          hookLog.push('preSubscriptionExecution');
        },
      },
    }),
  ],
  providers: [MercuriusReviewsResolver],
})
export class MercuriusSubscriptionsModule {}

import { CanActivate, ExecutionContext, Injectable } from '@nestjs/common';
import { GqlExecutionContext } from '@nestjs/graphql';
import { GraphQLError } from 'graphql';

@Injectable()
export class GqlAuthGuard implements CanActivate {
  canActivate(context: ExecutionContext): boolean {
    // Works for queries, mutations and subscriptions: the user is put on the
    // GraphQL context by the `context` option, never read from `req` here.
    const { user } = GqlExecutionContext.create(context).getContext();
    if (!user) {
      throw new GraphQLError('You must be signed in', {
        extensions: { code: 'UNAUTHENTICATED' },
      });
    }
    return true;
  }
}

curl -s localhost:3000/graphql -H 'content-type: application/json' \
  -d '{"query":"{ reviewsFor(bookId: 1) { id } }"}'

{ "errors": [ { "message": "You must be signed in", "locations": [ { "line": 1, "column": 3 } ], "path": [ "reviewsFor" ], "extensions": { "code": "UNAUTHENTICATED" } } ], "data": null }

import { createParamDecorator, ExecutionContext } from '@nestjs/common';
import { GqlExecutionContext } from '@nestjs/graphql';

export const CurrentUser = createParamDecorator((_data: unknown, context: ExecutionContext) =>
  GqlExecutionContext.create(context).getContext().user,
);

import { CallHandler, ExecutionContext, Injectable, NestInterceptor } from '@nestjs/common';
import { GqlExecutionContext } from '@nestjs/graphql';
import { Observable, tap } from 'rxjs';

export const timings: string[] = [];

@Injectable()
export class TimingInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<unknown> {
    const info = GqlExecutionContext.create(context).getInfo();
    const label = `${info.parentType.name}.${info.fieldName}`;
    const started = performance.now();
    return next.handle().pipe(
      tap(() => timings.push(`${label} ${(performance.now() - started).toFixed(1)}ms`)),
    );
  }
}

import { ArgumentsHost, Catch } from '@nestjs/common';
import { GqlArgumentsHost, GqlExceptionFilter } from '@nestjs/graphql';
import { GraphQLError } from 'graphql';

export class ReviewTooLongError extends Error {
  constructor(readonly maxLength: number) {
    super(`A review may have at most ${maxLength} characters`);
  }
}

@Catch(ReviewTooLongError)
export class ReviewTooLongFilter implements GqlExceptionFilter {
  catch(exception: ReviewTooLongError, host: ArgumentsHost) {
    const info = GqlArgumentsHost.create(host).getInfo();
    // Returning (or throwing) a GraphQLError controls the `errors` entry.
    return new GraphQLError(exception.message, {
      extensions: {
        code: 'REVIEW_TOO_LONG',
        maxLength: exception.maxLength,
        field: info.fieldName,
      },
    });
  }
}

{ "errors": [ { "message": "A review may have at most 200 characters", "locations": [ { "line": 1, "column": 12 } ], "path": [ "addReview" ], "extensions": { "code": "REVIEW_TOO_LONG", "maxLength": 200, "field": "addReview" } } ], "data": null }