Cloudflare Workers Topic-based GraphQL Subscriptions

This library uses Cloudflare Workers, Durable Objects and D1 to provide powerful topic-based GraphQL subscriptions.

Features:

• 👀 Easy to integrate with your existing GraphQL stack
• 🙌 Almost no setup
• 🔍 In-database JSON filtering
• 🗺 Publish from anywhere
• 🎹 Great typings
• 🔐 Authentication

// app.ts
import { makeExecutableSchema } from "@graphql-tools/schema";
import { createYoga } from "graphql-yoga";
import {
  handleSubscriptions,
  createWsConnectionPoolClass,
  subscribe,
  DefaultPublishableContext,
  createDefaultPublishableContext,
} from "graphql-workers-subscriptions";

export interface ENV {
  WS_CONNECTION_POOL: DurableObjectNamespace;
  SUBSCRIPTIONS: D1Database;
}

export const schema = makeExecutableSchema<DefaultPublishableContext<ENV>>({
  typeDefs: /* GraphQL */ `
    type Greeting {
      greeting: String
    }
    type Query {
      ping: String
    }
    type Subscription {
      greetings(greeting: String): Greeting
    }
    type Mutation {
      greet(greeting: String!): String
    }
  `,
  resolvers: {
    Query: {
      ping: () => "pong",
    },
    Mutation: {
      greet: async (root, args, context, info) => {
        context.publish("GREETINGS", {
          greetings: { greeting: args.greeting },
        });
        return "ok";
      },
    },
    Subscription: {
      greetings: {
        subscribe: subscribe("GREETINGS", {
          filter: (root, args, context, info) => {
            return args.greeting
              ? { greetings: { greeting: args.greeting } }
              : {};
          },
        }),
      },
    },
  },
});

const settings = {
  schema,
  wsConnectionPool: (env: ENV) => env.WS_CONNECTION_POOL,
  subscriptionsDb: (env: ENV) => env.SUBSCRIPTIONS,
};

const yoga = createYoga<DefaultPublishableContext<ENV>>({
  schema,
  graphiql: {
    // Use WebSockets in GraphiQL
    subscriptionsProtocol: "WS",
  },
});

const baseFetch: ExportedHandlerFetchHandler<ENV> = (
  request,
  env,
  executionCtx
) =>
  yoga.handleRequest(
    request,
    createDefaultPublishableContext({
      env,
      executionCtx,
      ...settings,
    })
  );

const fetch = handleSubscriptions({
  fetch: baseFetch,
  ...settings,
});

export default { fetch };

export const WsConnectionPool = createWsConnectionPoolClass(settings);

# wrangler.toml
[[migrations]]
new_classes = ["WsConnectionPool"]
tag = "v1"

[build]
# your build script
command = 'npm run build'

[[d1_databases]]
binding = "SUBSCRIPTIONS"
database_id = "877f1123-088e-43ed-8d4d-37e71c77157c"
database_name = "SUBSCRIPTIONS"
migrations_dir = "node_modules/graphql-workers-subscriptions/migrations"
preview_database_id = "877f1123-088e-43ed-8d4d-37e71c77157c"

[durable_objects]
bindings = [{name = "WS_CONNECTION_POOL", class_name = "WsConnectionPool"}]

Deployment

# create db
wrangler d1 create SUBSCRIPTIONS
# apply migrations
wrangler d1 migrations apply SUBSCRIPTIONS
# publish
wrangler publish

Local development

# create db
wrangler d1 create SUBSCRIPTIONS --local
# apply migrations
wrangler d1 migrations apply SUBSCRIPTIONS --local
# publish
wrangler dev

Authentication

For WebSocket authentication, you can pass onConnect to createWsConnectionPoolClass (read more documentation here):

export const WsConnectionPool = createWsConnectionPoolClass<ENV, { token: string }>({
  ...settings,
  onConnect: (ctx) => {
    const token = ctx.connectionParams.token;
    return isTokenValid(token, ctx.extra.env.TOKEN);
  },
});

Then, on the frontend:

import { GraphQLWsLink } from "@apollo/client/link/subscriptions";
import { createClient } from 'graphql-ws';

const wsLink = new GraphQLWsLink(
  createClient({
    url: "wss://subscriptions.workers.dev",
    connectionParams: () => {
      return { token: "<TOKEN>" };
    }
  })
);

If you want to verify individual subscriptions, use onSubscribe.

To verify the incoming HTTP requests, use isPublishAuthorized and isConnectAuthorized.

Publishing from outside Cloudflare

You can use POST /publish on your Worker to publish events.

curl -X POST https://graphql-workers-subscriptions.bubblydoo.workers.dev/publish -H 'Content-Type: application/json' -d '{"topic": "GREETINGS", "payload":{"greetings": {"greeting": "hi!"}}}'

To disable this, pass isPublishAuthorized: () => false to handleSubscriptions, or add custom authorization logic there.

Pooling

To minimize Durable Objects costs, one Pool can manage many WebSocket connections.

There are 3 pooling options:

• global: There is 1 global Pool that manages all WebSocket connections.
• colo: There's 1 Pool per datacenter, based on request.cf.colo
• continent: There's 1 Pool per continent, based on request.cf.continent (default)
• none: Every connection uses its own Pool (its own Durable Object)

You can also pass a custom pooling function to handleSubscriptions.

Expected costs

Considering you used up all your included credits in the Paid Plan:

One Pool (one Durable Object) is always charged at 128MB and will not sleep unless all its WebSockets are closed. Having one Pool run for 1 month (2600000 seconds) will be about 325000GBs, which would (at the time of writing) cost $4.0625. With a global pool, that would also be your maximum cost.

1 new connection causes:

• 1 D1 query
• 1 Durable Object request (fetch)

1 GraphQL subscription causes:

• 1 Durable Object fetch (incoming WebSocket message)

1 publish causes:

• 1 D1 query
• 1 Durable Object request (fetch) per Pool that has a connection that the publish will send to

If you would publish a message every second for 1 month to 100 WebSockets in 1 Pool, this would cause only 1 D1 query and 1 Durable Object request per second, and would cost you about $0.648/month. D1 is still free for now.

According to the pricing docs, there is no charge for outgoing WebSocket messages.

Internal details

Subscriptions are stored inside D1.

The D1 database has 6 columns:

• id (websocket message id that will be matched with the client-side subscription, a string, generated by the client)
• connectionId (a uuid, a string, identifying the WebSocket connection, generated by the Worker)
• connectionPoolId (a Durable Object id identifying the pool, a string, e.g. global, or US)
• subscription (the query the subscriber has requested, a JSON string)
• topic (a string)
• filter (the filter against which payloads are checked, a JSON string or null)

The Durable Object has a reference to the WebSocket, which can then be used to publish data to.

Filters are compared in-database using:

SELECT * FROM Subscriptions WHERE topic = ?1 AND (filter is null OR json_patch(?2, filter) = ?2);

with ?1: topic and ?2: payload.

Contributing

Check out this repo, then run:

yarn

yarn build

Then link the package to your project (you can take the example as start).

Bundling issue

Due to the dual package hazard in GraphQL (see this issue) you might get duplicate "graphql" modules cannot be used at the same time errors.

This is because both the CJS and ESM version of graphql are loaded.

In that case, you might have to bundle yourself. When using esbuild, the option --resolve-extensions=.mts,.mjs,.ts,.js,.json works. See the build-app script in package.json for an example.

Credits 🙏

This project was inspired by cloudflare-worker-graphql-ws-template and subscriptionless.