Is there an example for GraphQL Subscriptions?

[lukaskemper]

Hello,
I'm building an NestJs GraphQL API with subscriptions. This API has a complex middleware to ensure, that user are getting only the data that they allowed to.

Now I'm trying to implement subscriptions endpoints with filter method like here. This filter method does a database call via Prisma. The Prisma middleware is triggered and manipulates the database query to ensure the listener gets only a notification for events he's allowed to. The middleware uses the NestJs-CLS context to identify the user and manipulate the database query correctly.

Now I'm facing the problem, that the mutation request which will publish an event overrides the NestJs-CLS context so that the Prisma middleware (which is triggered from the subscription filter method) does not have any context (here I need the context of the user, which listened to the subscription).

Here is a code example:

@Subscription(() => Approval, {
    async filter(
      this: ApprovalSubResolver,
      payload: { onFileApproved: Approval },
      _: undefined,
      context,
    ): Promise<boolean> {
      return true;
      // try {
      //   return !!(await this.approvalService.findOneApproval( <-- This will trigger the middleware
      //     {
      //       id: payload.onFileApproved.id,
      //     },
      //     undefined,
      //     true,
      //   ));
      // } catch (e) {
      //   console.error(e);
      // }
    },
  })
  async onFileApproved() {
    console.log(this.clsService.get()); // Logs correct context of websocket connection
    this.clsService.runWith<void>(this.clsService.get(), async () => {
      console.log('INSIDE RUN WITH', this.clsService.get()); // Logs correct context of websocket connection
      await this.pubSub.subscribe('onFileApproved', (res) => {
        console.log(this.clsService.get('user')); // Logs the context http request (mutation) where the message is published 
      });
    });
    // this.clsService.enterWith(this.clsService.get());
    return this.pubSub.asyncIterator<Approval>('onFileApproved');
  }

I'm confused.

[Papooch]

Hi, I have not used GrahQL subscriptions before, but it seems that since the middleware call is defined in the Subscription decorator, Nest calls the filter function internally outside of the whole call chain started with cls.run.

Are you using a middleware or an interceptor to set up the ClsContext? In case of an interceptor, you can leverage a little "workaround" I had to do to keep the CLS context the same across all parallel queries/mutations within a single request.

As a result of this, if you can get ahold of the ExecutionContext or the GqlContext objects in the fitler, see if you can get the ClsStore under the __CLS_STORE__ key on the graphql context:

context // ExecutionContext
  .getArgByIndex(2) // GqlContext (if context.getType() == 'graphql')
  .__CLS_STORE__ // the ClsStore

If that works, I can investigate how this can be turned into a public API and make it more stable.

[lukaskemper]

Hi, thank you for your fast reply.

I'm now getting the context correctly inside the filter method.

async filter(
      this: ApprovalResolver,
      payload: { onFileApproved: Approval },
      _: undefined,
      context,
    ): Promise<boolean> {
      console.log(context.__CLS_STORE__); // Correct Context
      return true;
    },

But now I'm wondering how I can access this Context inside my Prisma middleware.
The filter method should look like this:

async filter(
      this: ApprovalResolver,
      payload: { onFileApproved: Approval },
      _: undefined,
      context,
    ): Promise<boolean> {
         return !!(await this.approvalService.findOneApproval(
           {
             id: payload.onFileApproved.id,
           },
           undefined,
           true,
         ));
    },

The method findOneApproval on approvalService will trigger my Prisma middleware.

@Injectable()
export class PrismaService
  extends PrismaClient<Prisma.PrismaClientOptions, 'query' | 'error'>
  implements OnModuleInit
{
  private readonly logger = new Logger(PrismaService.name);
  constructor(private readonly config: ConfigService) {
    super({
      log: [
        { emit: 'stdout', level: 'info' },
        { emit: 'stdout', level: 'warn' },
        { emit: 'event', level: 'query' },
      ],
      errorFormat: 'minimal',
    });
    this.$use(prismaMiddleware()); // <-- This function is called on every Prisma operation.
  }

  async onModuleInit(): Promise<void> {
    if (this.config.get<string>('NODE_ENV') === 'local') {
      this.$on('query', (event) => {
        this.logger.debug(`${event.query} +${event.duration}ms`);
      });
    }
    await this.$connect();
  }

  async enableShutdownHooks(app: INestApplication): Promise<void> {
    this.$on('beforeExit', async () => {
      await app.close();
    });
  }
}

export function prismaMiddleware<
  T extends Prisma.BatchPayload = Prisma.BatchPayload,
>(): Prisma.Middleware {
  return async (
    params: Prisma.MiddlewareParams,
    next: (params: Prisma.MiddlewareParams) => Promise<T>,
  ): Promise<T | GraphQLError> => {
    const cls = ClsServiceManager.getClsService<IClsStore>();

   console.log(cls.get('user')); // <-- Is undefined
    const requesterId = cls.get('user.sub');
}

At the primsaMiddleware function I'm using the ClsService to manipulate the database queries based on value of the context. How can I access the context.__CLS_STORE__ inside the prismaMiddleware function?

This middleware is also used by normal mutations and queries where everything works finde, also inside the middleware.

[Papooch]

If the decorator function has the same this context as the class, you can just use the already injected ClsService, if not, you'll need to use the static ClsServiceManager to retrieve the singleton ClsService instance.

Then just wrap the middleware call and pass in the store:

return await this.clsService.runWith(context.__CLS_STORE__, () => {
  return // ... call the middleware
})

The context will be then preserved within this callback/promise chain.

(remember that cls.run/With returns whatever the callback returns)

[lukaskemper]

Oh yes, that makes sense... I tried soooo much 😄

Thank you very much! You saved my day!

PS: Your library is amazing, keep it on!

[Papooch]

@lodeli-lulu
Please be aware that I am changing some internal implementation in a fix which will remove this "feature" that you're relying on. The change will happen in a patch version from 4.2.0 to 4.2.1 (as it fixes other broken behavior) and it will remove the __CLS_STORE__ "escape hatch".

Since version 4.2.1, it will now be possible to retrieve the ClsStore instance during a request with ContextClsStoreMap.getByRaw(gqlContext), so you will need to change your code accordingly otherwise it will no longer work as intended:

const existingStore = ContextClsStoreMap.getByRaw(context)
return await this.clsService.runWith(existingStore, () => {
  return // ... call the middleware
})

The ContextClsStoreMap is now a part of a public API, so it should be more stable than relying on magic properties attached to the context, but I'm still leaving it intentionally undocumented.

[lukaskemper]

Hey, thank you for the note. I really appreciate it.