Subscriptions: New Implementation Proposal

[arthursimas1]

Feature Request

The current Subscription implementation requires the registration of a PubSub system even when it's not used.

Minimal example

import 'reflect-metadata'

import { createServer } from 'node:http'
import { createYoga } from 'graphql-yoga'
import { buildSchemaSync, Resolver, Query, Subscription, Arg, Root } from 'type-graphql'

@Resolver()
class SampleResolver {
  @Query(returns => Boolean)
  sample() {
    return true
  }

  @Subscription(returns => Number, {
    subscribe: async function *({ args }) {
      for (let i = args.from; i >= 0; i--) {
        await new Promise(resolve => setTimeout(resolve, 1000))
        yield i
      }
    }
  })
  countdown(
    @Root() payload: Number,
    @Arg('from', type => Number) from: number,
  ) {
    return payload
  }
}

function bootstrap() {
  const schema = buildSchemaSync({
    resolvers: [
      SampleResolver,
    ],
    pubSub: true as any,
  })

  const yoga = createYoga({ schema })
  const server = createServer(yoga)

  server.listen(4000, () => {
    console.info('Server is running on http://localhost:4000/graphql')
  })
}
bootstrap()

Some considerations that deserve attention are:

1. Notice the pubSub parameter. In this case, it's totally unused and ignored internally. We can be sure of it by the fact that it's just a truthy value; it doesn't even comply with the PubSub interface
2. In subscribe, args.from is not typed. If we want just a bit more type-safety, we could use { args }: { args: { from: number } }, but nothing protects us from changing @Arg('from', type => Number) from: number and face terrible errors
3. In countdown, @Arg('from', type => Number) from: number provides the type safety we need, but in the wrong place and is totally unused in countdown
4. We're unable to use dependency injection in the subscribe function. I've came up with a workaround to e.g. access the database, but it's too ugly, not type-safe and error prone

In this example, the underlying PubSub System is unused, but when it is, it's just to handle topics and common PubSub operations.

Proposal I propose to change the Subscription decorator implementation to allow the decorated function to be a Generator or return a Generator, so it can allow more use cases, even the current supported ones.

With that change, the code becomes simpler and easier to read:

@Resolver()
class SampleResolver {
  // ...

  @Subscription(returns => Number)
  async *countdown(
    @Arg('from', type => Number) from: number,
  ) {
    for (let i = from; i >= 0; i--) {
      await new Promise(resolve => setTimeout(resolve, 1000))
      yield i
    }
  }
}

And when a PubSub System is required, it's simple to add:

import { createPubSub } from '@graphql-yoga/subscription'

const pubSub = createPubSub()

@Resolver()
class SampleResolver {
  // ...

  @Mutation(returns => Boolean)
  publish(
    @Arg('topic', type => String) topic: string,
    @Arg('data', type => String) data: string,
  ) {
    pubSub.publish(topic, data)
    return true
  }

  @Subscription(returns => String)
  subscribe(
    @Arg('topic', type => String) topic: string,
  ) {
    return pubSub.subscribe(topic)
  }
}

A parameter can be used to signal if the function is a Generator or if it returns a Generator, or even inference it somehow (maybe Symbol.iterator and Symbol.asyncIterator can help).

---

This could be implemented as (1) a hard breaking change, (2) flag the old method as deprecated in the meantime, or (3) support both methods.

I know that Subscriptions are a complicated thing that this library has already worked a lot in the past releases, so I could help submitting a PR to support this change.

[MichalLytek]

We have a proposal for a more refined subscriptions system in #542. It would let you define all those stuff we do in decorator options as a method parameter.

Please check it out and let me know if you find your proposal fits in that issue and maybe put a comment about no pubsub but custom async generator as subscribe method in ADR.

[arthursimas1]

I don't think the proposal in #542 fits the code pattern that this library implicitly imposes.

What I mean is, when a developer first sees a codebase using type-graphql, they can immediately understand how to write a query and a mutation. These operations are almost the same in terms of implementation.

But when it comes to a subscription, in the current way, it changes a little bit, as one would need to use a PubSub System or implement a custom Generator for the subscribe parameter in @Subscription (and facing the inherent limitations of it).

As far as I could notice, #542 changes even further, proposing the ADR pattern, which is totally different of how queries and mutations are currently implemented.

In the ideal world, I think that all operations (query, mutation and subscription) could be implemented in the same way, without big changes. If someday the library supports writing the subscription operation using the ADR pattern, I think both query and mutation would also need to support it.

This is my current opinion regarding it. What do you think?

[arthursimas1]

I've edited the issue description with a fourth problem of the current implementation (which I forgot to initially mention).

[arthursimas1]

I feel that #617 can also be resolved by my proposal.

[MichalLytek]

What do you think?

I think that queries/mutations and subscriptions are so much different that they deserve a different treatment in terms of API. Trying to add subscribe, filter, topics, etc. in decorators have it's own limitation that you've mentioned.

With #542 being implemented, what would be still not possible that your proposal #1638 solves?

[arthursimas1]

1.

I think that queries/mutations and subscriptions are so much different that they deserve a different treatment in terms of API. Trying to add subscribe, filter, topics, etc. in decorators have it's own limitation that you've mentioned.

When you move filters, topics, and other subscription-related things to the body of the function, instead of using decorators, the subscription can be more "configurable", making the three operations look similar.

There's a lot of cool libraries that can be used to implement the subscription mechanism (@graphql-yoga/subscription, apollographql/graphql-subscriptions, Repeater.js, event-iterator). They're framework-agnostic. Given so, type-graphql don't need to implement nothing special to support them. Subscriptions can be handled by the library the user wishes, not necessarily built in type-graphql. We have the benefit to reuse third-party libraries for a lot of things, like using graphql-scalars as possible instead of implementing our own scalars everytime. It can be the case for subscriptions.

Doing so, authentication, middlewares, and everything else we can already do with queries and mutations, could be done with subscriptions, without requiring a different API, as it would be the same flow.

2.

With https://github.com/MichalLytek/type-graphql/issues/542 being implemented, what would be still not possible that your proposal https://github.com/MichalLytek/type-graphql/issues/1638 solves?

Simplicity and same API for query, mutation and subscription.

In #542, subscribe returns AsyncIterator | Promise<AsyncIterator>, it's the same return type as mine. So it introduces a different API for the same thing, that can be done in a simpler way (as far as I could understand the proposal).

Can you please provide a more detailed example using that pattern, so we can understand better the proposal? i.e. use the ADR pattern to implement the SampleResolver I've used. Where's how a sample code would look like with my proposal:

import { Repeater, createPubSub, filter, map, pipe } from '@graphql-yoga/subscription'
import { Resolver, Query, Mutation, Subscription, Arg } from 'type-graphql'

enum TOPICS {
  INTERNAL_status = 'INTERNAL_status',
  status = 'status',
}

const pubSub = createPubSub<{
  [TOPICS.status]: [string],
  [TOPICS.INTERNAL_status]: [string],
}>()

@Resolver()
class SampleResolver {
  constructor(
    private statusService: StatusService,
  ) {}

  @Subscription(returns => Number)
  async *countdown(
    @Arg('from', type => Number) from: number,
  ): AsyncIterator<number> {
    for (let i = from; i >= 0; i--) {
      await new Promise(resolve => setTimeout(resolve, 1000))
      yield i
    }
  }

  @Subscription(returns => Number)
  countdownRepeater(
    @Arg('from', type => Number) from: number,
  ): AsyncIterator<number> {
    return new Repeater<number>(async (push, stop) => {
      for (let i = from; i >= 0; i--) {
        await new Promise(resolve => setTimeout(resolve, 1000))
        await push(i)
      }
      stop()
    })
  }

  @Mutation(returns => Boolean)
  updateInternalStatus(
    @Arg('data', type => String) data: string,
  ) {
    pubSub.publish(TOPICS.INTERNAL_status, data)
    return true
  }

  @Mutation(returns => Boolean)
  updateStatus(
    @Arg('data', type => String) data: string,
  ) {
    pubSub.publish(TOPICS.status, data)
    return true
  }

  @Subscription(returns => String)
  subscribeToStatusChanges(
    @Arg('exclude', type => [String]) exclude: string[],
  ): AsyncIterator<string> {
    // create a subscription pipeline
    return pipe(
      // subscribe to multiple topics and sources
      Repeater.merge([
        pubSub.subscribe(TOPICS.status),
        pubSub.subscribe(TOPICS.INTERNAL_status),
        this.statusService.subscribe(),
      ]),

      // map values
      map((value) => value.toLowerCase()),

      // filter values
      filter((value) => !exclude.includes(value)),
    )
  }
}

Currently I'm unable to financially support this project. Helping it with some enhancements is my way to give back to the open-source community. In the meantime ADR isn't implemented, I could implement my proposal and submit a PR so you can check it out without commitments.

[MichalLytek]

After seeing #1641, I can answer your points:

Ad 1. I agree, we can defer checking for pubsub and throw error only if no subscribe parameter is applied. Like you did in your PR #1641.

Ad 4. You can put your container in the graphql context (if not shared) and use it as a service locator to get an instance with resolved dependencies, e.g.: ({ context }) => context.container.resolve(DbSubscriptionHandler).subscribe(context.user.id);

Ad 2. and 3. The #542 would solve that issue by allowing to define those subscribe and resolve steps separately, with merging of the @Arg declaration in each method. So that you can have part of the args used to subscribe, to filter and to resolve, depending on your needs.

Doing so, authentication, middlewares, and everything else we can already do with queries and mutations, could be done with subscriptions, without requiring a different API, as it would be the same flow.

Your proposal flips the standard TypeGraphQL apporach of class methods being the services that gets inputs and returns values. In your case, the method returns async iterator, and the resolve part has to be implemented manually. So all the middlewares and other stuff does not work on resolve level, you can't log each resolve call, only once on subscription, etc.

So while we both agree that GraphQL subscriptions are much more complex than queries and mutations, and we need to change the API to have a better DX, we have different opinions about the desired approach. You want to flip the convention and allow subscribe instead of resolve in the method body. I adhere to the #542 and having subscriptions as classes with steps/phases as methods.

So to sum up: I don't find your proposal being replacement for #542 as it expose complex internals of repeater.js and other as a first class citizen in order to fix issue from 2. and 3. that #542 would solve in a much cleaner way. So while I'm grateful for your contribution with this feature request and even implementation in #1641, unfortunately I have to reject this proposal. Feel free to publish your changes as a fork on npm, until the #542 gets implemented and you could transition to the official version if you want 😉