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

``ts
// 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>({
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>({
schema,
graphiql: {
// Use WebSockets in GraphiQL
subscriptionsProtocol: "WS",
},
});

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

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

export default { fetch };

export const WsConnectionPool = createWsConnectionPoolClass(settings);
`

`toml


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"}]
`

$3

`shell


create db

wrangler d1 create SUBSCRIPTIONS

apply migrations

wrangler d1 migrations apply SUBSCRIPTIONS

publish

wrangler publish
`

$3

`shell


create db

wrangler d1 create SUBSCRIPTIONS --local

apply migrations

wrangler d1 migrations apply SUBSCRIPTIONS --local

publish

wrangler dev
`

$3

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

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

Then, on the frontend:

`ts
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: "" };
}
})
);
`

If you want to verify individual subscriptions, use onSubscribe.

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

$3

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

`shell
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.

$3

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.

$3

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.

$3

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:

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

with ?1: topic and ?2: payload.

$3

Check out this repo, then run:

`shell
yarn

yarn build
`

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

$3

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.

$3

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