YogaDriver for NestJS: support nestjs/graphql's @Subscription filter function

[gthau]

PR submitted: #2977

Is your feature request related to a problem? Please describe.
NestJS GraphQL declares Subscriptions' resolvers using the @Subscription decorator. This decorator accepts an options object with a resolve and filter functions. Currently the YogaDriver for NestJS doesn't support the filter method.
It should be supported to ensure feature parity with the Apollo and Mercurius driver and to make sure the NestJS main documentation is relevant to devs choosing to use the YogaDriver.

Describe the solution you'd like
Add support for the filter function.

@Subscription('filteredGreetings', {
  filter: (payload, variables) => {
    return payload.filteredGreetings
      .toLowerCase()
      .startsWith(variables.firstLetter.toLowerCase());
  },
})
async *filteredGreetings() {
  for (const hi of ['Hi', 'Bonjour', 'Hola', 'Ciao', 'Zdravo']) {
    yield { filteredGreetings: hi };
  }
}

Describe alternatives you've considered
Two alternatives:

1. Filter the async iterable in the resolver, for example using Yoga's own utils pipe and filter

@Subscription('filteredGreetings')
async *filteredGreetings() {
  return pipe(
    pubSub.subscribe('greetings'),
    filter((payload) => payload
      .filteredGreetings.toLowerCase()
      .startsWith(variables.firstLetter.toLowerCase()),
  );
}

2. Subclass the YogaDriver class to add the missing method

export class PatchedYogaDriver extends YogaDriver {
  public subscriptionWithFilter(
    instanceRef: unknown,
    filterFn: (
      payload: any,
      variables: any,
      context: any,
    ) => boolean | Promise<boolean>,
    // we don't care about the specific signature of the createSubscribeContext function
    // eslint-disable-next-line @typescript-eslint/ban-types
    createSubscribeContext: Function,
  ) {
    return async <TPayload, TVariables, TContext, TInfo>(
      ...args: [TPayload, TVariables, TContext, TInfo]
    ): Promise<any> => {
      return pipe(
        await createSubscribeContext()(...args),
        filter((payload: any) =>
          filterFn.call(instanceRef, payload, ...args.slice(1)),
        ),
      );
    };
  }
}

@Module({
  imports: [
    GraphQLModule.forRoot<YogaDriverConfig>({
      driver: PatchedYogaDriver, // <--- use patched Yoga Driver
      // ...
  ],
  controllers: [AppController],
})
export class AppModule {}

[bernessco]

Hey!
The fix looks promising, but there is still a problem. If you are using PubSub from graphql-subscriptions library, instead of Yoga's built-in one you are getting an error while trying to subscribe to a filtered subscription

Subscription field must return Async Iterable. Received: [function piped].

[bernessco]

I guess this is expected, but if we just add this implementation from apollo driver and copy createAsyncIterator it will be possible to make it work with some tweaking or config options both ways

public subscriptionWithFilter(
    instanceRef: unknown,
    filterFn: (
      payload: any,
      variables: any,
      context: any,
    ) => boolean | Promise<boolean>,
    createSubscribeContext: Function,
  ) {
    return <TPayload, TVariables, TContext, TInfo>(
      ...args: [TPayload, TVariables, TContext, TInfo]
    ): any =>
      createAsyncIterator(createSubscribeContext()(...args), (payload: any) =>
        filterFn.call(instanceRef, payload, ...args.slice(1) as [TVariables, TContext]),
      );
  }

[gthau]

xpected, but if we just add this implementation from apollo driver an

Hi Mantas,
are you sure you are returning an AsyncIterator from the resolver, and not just the PubSub?
The graphql-subscriptions PubSub class has an .asyncIterator method which needs to be called.
This is working for me:

import { PubSub } from 'graphql-subscriptions';

@Resolver()
export class TestSubscriptionsResolver {
  private readonly pubSub = new PubSub();
  
  constructor() {
    setInterval(() => {
      this.pubSub.publish('now', Math.floor(Date.now() / 1_000));
    }, 1000);
  }
  @Subscription(() => TestSubscriptionOutput, {
    filter: (payload) => payload % 2 === 0,
    resolve: (payload) => ({ countdown: payload }),
  })
  async testSubscription(@Args('input') input: number) {
    return this.pubSub.asyncIterator('now');
  }
}

[bernessco]

Yeah it works with the snippet that i sent from apollo

export const OrderUpdatedSubscriptionName = 'orderUpdated'

export interface OrderUpdatedSubscriptionPayload {
    [OrderUpdatedSubscriptionName]: OrderUpdatedEventPayload;
}

@Subscription(() => OrderUpdatedPayload, {
         filter: ({ orderUpdated }: OrderUpdatedSubscriptionPayload, _, { req: { user } }) => {
            console.log(11111, orderUpdated.order)
            return orderUpdated.order.userId === user.id
        },
        resolve: payload => {
            console.log(2222, payload)
            return payload
        }
    })
    async [OrderUpdatedSubscriptionName] () {
        return this.pubSubService.asyncIterator(OrderUpdatedSubscriptionName)
    }

// Publishing looks like
this.pubSubService.publish<OrderUpdatedSubscriptionPayload>(OrderUpdatedSubscriptionName, {
        [OrderUpdatedSubscriptionName]: event.payload
    })

[bernessco]

Now that Im looking at the code, maybe its related to the returned payload format
https://docs.nestjs.com/graphql/subscriptions#publishing

This tells us that the subscription must return an object with a top-level property name of commentAdded that has a value which is a Comment object. The important point to note is that the shape of the event payload emitted by the PubSub#publish method must correspond to the shape of the value expected to return from the subscription. So, in our example above, the pubSub.publish('commentAdded', { commentAdded: newComment }) statement publishes a commentAdded event with the appropriately shaped payload. If these shapes don't match, your subscription will fail during the GraphQL validation phase.

[gthau]

Yes, what you take as input of filter and resolve matches the return of the Subscription method, then what you return from the resolve function must match the expected input type of the next type resolver.

Same as in envelop/Yoga when you use query/mutation/subscription resolvers, except that codegen mappers gives you type safety through the use of the generated resolvers' signatures, but with NestJS you don't get that because everything is based on decorators.

[bernessco]

Yeah, so by documentation:

If you use the resolve option, you should return the unwrapped payload (e.g., with our example, return a newComment object directly, not a { commentAdded: newComment } object).

So this patch should work with structure that i provided, because thats what graphql-subscription expects. Do you think we can find a solution to support both built in and graphql-subscription?

[gthau]

It already works, it's agnostic to graphql-subscriptions PubSub class.
What you get as input of filter is the output of the method decorator by @Subscription.
In your case the issue is that you "double" wrap your payload by publishing a named event AND wrapping the published payload.

this.pubSubService.publish<OrderUpdatedSubscriptionPayload>(OrderUpdatedSubscriptionName, {
    [OrderUpdatedSubscriptionName]: event.payload
})

What you should do is

this.pubSubService.publish<OrderUpdatedSubscriptionPayload>(OrderUpdatedSubscriptionName, event.payload);

The payload is already identified by the name of the event used to publish, subscribe and get an asyncIterator.

[bernessco]

Yeah i got this before, while analysing the error, but the problem is that if we make this then it goes agains nestjs documentation as it says to structure the payload nested that way

https://docs.nestjs.com/graphql/subscriptions#publishing

[bernessco]

@gthau Lol, i think i figured it out... I was using pipe, filter from rxjs instead of yoga...
Got to the root, basically if you don't want to use resolve on every subscription this structure is for that, to auto resolve.
Btw everything works as expected both ways now, so this is valid fix

[gthau]

Yeah i got this before, while analysing the error, but the problem is that if we make this then it goes agains nestjs documentation as it says to structure the payload nested that way

https://docs.nestjs.com/graphql/subscriptions#publishing

Imo this part of the documentation is bad. It mixes everything with no respect for separation of concerns. It doesn't make sense for the mutation to need to know about the subscription name in order to wrap the event, that's contrary to the goal of using a pubsub/bus/etc. Moreover the event might be used for several subscriptions, etc.

Also it confuses about how the resolvers' chain works. Returning exactly the exact shape from the operation resolver completely ignores the fact that type resolvers might be used to recursively produce the correct shape.

But it's not the topic of this PR 😄

[enisdenjo]

This issue was resolved with #2977 and released with in @graphql-yoga/nestjs@2.1.0 and @graphql-yoga/nestjs-federation@2.2.0.