Smart subscriptions plugin

Turn any query into a live GraphQL subscription by registering events on the fields and object types it touches.

A smart subscription runs a normal query, pushes the first result to the client, and then keeps that result live. Every field and object type the query touches can register the events that should refresh it, and when one fires the plugin re-runs the query and pushes the new data. You opt a query field into this with smartSubscription: true, and you register events from a subscribe callback on any field or object type.

The flow has three steps:

Run the query the subscription is based on and push its initial result to the client.
As the query resolves, register any subscriptions declared on the fields and object types it used.
When a registered event fires, re-run the query and push the updated data.

You never decide up front which parts of the schema are subscribable. Any type or field can register an event, so the same query can drive a live view of a whole game or of a single score line. Refetch options (below) narrow a refresh to just the sub-tree that changed, so a single score update doesn't re-run the whole query.

Install

npm install --save @pothos/plugin-smart-subscriptions

Setup

Register the plugin and supply a smartSubscriptions object. Its subscribe and unsubscribe functions are how the plugin attaches to your event source: a pub/sub bus, a Redis channel, anything that can call back when an event with a given name fires.

import SchemaBuilder from '@pothos/core';
import SmartSubscriptionsPlugin from '@pothos/plugin-smart-subscriptions';

const builder = new SchemaBuilder<{ Context: Context }>({
  plugins: [SmartSubscriptionsPlugin],
  smartSubscriptions: {
    // Debouncing toggle: pass null to disable it, any non-null value to enable a short debounce window.
    debounceDelay: 10,
    subscribe: (name, context, cb) => context.pubsub.subscribe(name, cb),
    unsubscribe: (name, context) => context.pubsub.unsubscribe(name),
  },
});

Option	Purpose
`subscribe`	`(name, context, cb) => Promise<void> \| void`. Start listening for `name`; call `cb(err, data)` on each event.
`unsubscribe`	`(name, context) => Promise<void> \| void`. Stop listening for `name`.
`debounceDelay`	Toggles debouncing of event bursts before re-running the query. Pass `null` to disable it; any non-null value enables a short debounce window.

Wiring an async iterator

Most pub/sub libraries expose an async iterator per channel instead of a callback. subscribeOptionsFromIterator adapts one into the subscribe/unsubscribe pair for you:

import SchemaBuilder from '@pothos/core';
import SmartSubscriptionsPlugin, {
  subscribeOptionsFromIterator,
} from '@pothos/plugin-smart-subscriptions';

const builder = new SchemaBuilder<{ Context: Context }>({
  plugins: [SmartSubscriptionsPlugin],
  smartSubscriptions: {
    debounceDelay: 10,
    ...subscribeOptionsFromIterator((name, { pubsub }) =>
      pubsub.asyncIterableIterator(name),
    ),
  },
});

Creating a smart subscription

Add smartSubscription: true to a query field. The plugin mirrors it onto the schema's Subscription type under the same name, so Query.games gains a matching Subscription.games. The field's optional subscribe callback registers the events that should refresh the whole query:

builder.queryFields((t) => ({
  games: t.field({
    type: [Game],
    smartSubscription: true,
    subscribe: (subscriptions, root, args, ctx, info) => {
      subscriptions.register('game-added');
      subscriptions.register('game-removed');
    },
    resolve: (root, args, ctx) => ctx.Games.all(),
  }),
}));

Clients subscribe to it like any other subscription; the selection set is a normal query:

subscription {
  games {
    matchup
    scores {
      id
      points
    }
  }
}

Subscriptions on object types

An object type registers events with a subscribe option. It runs once for every instance of that type in the result, so a list of games registers one game/{id} event per game. When the query re-runs after an event, subscribe runs again for each object in the new result set.

Because subscribe is a standard object-type option, declare it right on the implement call for your objectRef:

const Game = builder.objectRef<IGame>('Game').implement({
  subscribe: (subscriptions, game, context) => {
    subscriptions.register(`game/${game.id}`);
  },
  fields: (t) => ({
    matchup: t.exposeString('matchup'),
    scores: t.field({ type: [Score], resolve: (game) => game.scores }),
  }),
});

register takes an options object as its second argument to control what a matched event does:

const Game = builder.objectRef<IGame>('Game').implement({
  subscribe: (subscriptions, game, context) => {
    subscriptions.register(`game/${game.id}`, {
      filter: (value) => value.gameId === game.id,
      invalidateCache: (value) => context.GameCache.remove(game.id),
      refetch: () => context.Games.fetchById(game.id),
    });
  },
  fields: (t) => ({
    matchup: t.exposeString('matchup'),
    scores: t.field({ type: [Score], resolve: (game) => game.scores }),
  }),
});

Option	Effect
`filter`	Called with the event value; the refresh only happens when it returns `true`.
`invalidateCache`	Called before refetching so you can clear stale cache entries, so the reload sees fresh data.
`refetch`	Refetch just this object. When provided, an event for this object (or any of its children) refreshes only this sub-tree; parts of the query that don't depend on it are left untouched.

Subscriptions on fields

When one field has a narrower refresh trigger than its parent type, register its events through the same subscribe callback in the field options:

const Game = builder.objectRef<IGame>('Game').implement({
  fields: (t) => ({
    matchup: t.exposeString('matchup'),
    scores: t.field({
      type: [Score],
      subscribe: (subscriptions, game) =>
        subscriptions.register(`game-scores/${game.id}`),
      resolve: (game) => game.scores,
    }),
  }),
});

Fields accept the same filter and invalidateCache options on register. In place of a refetch function, set canRefetch: true in the field options: the plugin re-runs this field's own resolver (and its children) instead of the rest of the query.

const Game = builder.objectRef<IGame>('Game').implement({
  fields: (t) => ({
    matchup: t.exposeString('matchup'),
    scores: t.field({
      type: [Score],
      canRefetch: true,
      subscribe: (subscriptions, game, args, context) =>
        subscriptions.register(`game-scores/${game.id}`, {
          filter: (value) => value.gameId === game.id,
          invalidateCache: () => context.GameCache.remove(game.id),
        }),
      resolve: (game) => game.scores,
    }),
  }),
});

Known limitations

The value passed to filter and invalidateCache is typed as unknown, so you'll narrow or cast it yourself. Smart subscriptions also don't work with list fields backed by async generators (the pattern behind @stream queries).

Smart subscriptions plugin

On this page