GraphQL Live Queries with Socket.io
Tags: graphql, javascript, node, webdev
Published: 2020-08-17 20:35:26

**Photo by <a href="https://unsplash.com/@nexgenfx">Luca Campioni</a> on <a href="https://unsplash.com/s/photos/speed">Unsplash</a>**

**Note:** For a better understanding I recommend reading [Subscriptions and Live Queries - Real-Time with GraphQL](https://the-guild.dev/blog/subscriptions-and-live-queries-real-time-with-graphql) first.

I have been using a GraphQL schema served over Socket.io for a long time. Finally, I abstracted the protocol for both client and server into a library for easier re-use across projects.

In addition to the usual GraphQL operations, I also added support for executing live queries.

**TL;DR:**

- **`@n1ru4l/socket-io-graphql-server`**: A layer for serving a GraphQL schema via a socket.io server. Supports Queries, Mutations, Subscriptions, and Live Queries.
- **`@n1ru4l/socket-io-graphql-client`**. A network interface for consuming a GraphQL schema that is served via `@n1ru4l/socket-io-graphql-server`. Can be used with all major GraphQL clients such as Relay, Apollo Client, or Urql.
- **`@n1ru4l/graphql-live-query`**. Utilities for adding live queries to **ANY** GraphQL schema.
- **`@n1ru4l/in-memory-live-query-store`**. GraphQL live query implementation.

All the packages can be found in this repository:

https://github.com/n1ru4l/graphql-live-queries

For showcasing the libraries I created a todo example app that syncs its state across all clients using the above packages:

- [**`Server`**](https://github.com/n1ru4l/graphql-live-queries/tree/master/packages/todo-example/server): todo app server implementation with graphql-js, `@n1ru4l/socket-io-graphql-server` and `@n1ru4l/in-memory-live-query-store`.
- [**`Client Relay`**](https://github.com/n1ru4l/graphql-live-queries/tree/master/packages/todo-example/client-relay): Todo app client implementation with `create-react-app`, `relay` and `@n1ru4l/socket-io-graphql-client`.
- [**`Client Apollo`**](https://github.com/n1ru4l/graphql-live-queries/tree/master/packages/todo-example/client-apollo): Todo app client implementation with `create-react-app`, `@apollo/client` and `@n1ru4l/socket-io-graphql-client`.
- [**`Client Urql`**](https://github.com/n1ru4l/graphql-live-queries/tree/master/packages/todo-example/client-urql): Todo app client implementation with `create-react-app`, `urql` and `@n1ru4l/socket-io-graphql-client`.

# Motivation

It seems like real-time for GraphQL is not pushed as much as it should by all the big players out there.

The most popular Node.js implementation for subscriptions is [poorly maintained as apollo focuses more on different fields](https://github.com/apollographql/graphql-subscriptions).

There a few live query implementations and **NONE** that are not tied to a specific database out there.

A few cool ideas popped up in the community (such as [`graphql-live-subscriptions`](https://github.com/D1plo1d/graphql-live-subscriptions). But none of those are maintained or have some major flaws such as not being compatible with interface or union types.

The implementations in `@n1ru4l/graphql-live-query` and `@n1ru4l/in-memory-live-query-store` should serve as an example of how it could be done without being tied to any specific (reactive) database or data structure. The implementation will hopefully mature with time as people report new use-cases and start adopting live queries.

In addition to that, I created both `@n1ru4l/socket-io-graphql-server` and `@n1ru4l/socket-io-graphql-client` as I was already heavily using GraphQL over Socket.io in a project that required real-time updates.

## GraphQL already has a solution for real-time

Subscriptions are the perfect tool for responding to events. An example of this would be triggering a sound or showing a toast message once a new message has been received.

Subscriptions are also often used for updating existing query results on the client. Depending on the complexity cache update code can eventually become pretty bloated. Often it is more straight-forward to simply refetch the query once a subscription event is received.

## Live Query Magic

Live queries however should feel magical and update the UI with the latest data from the server without having to do any cache update wizardry. This moves the complexity from the client to the server.

# Concept

The current definition of a live query for my project is a query operation that is annotated with a `@live` directive.

```graphql
query users @live {
  users(first: 10) {
    id
    login
  }
}
```

A live query is sent to the server (via WebSocket or HTTP) and stored there until the client disconnects or notifies the server he is no longer interested in the live query operation (and hence the server disposing of it).

On the server, the query is re-executed once the data selected by the live query operation changes. The result is then streamed to the client. A re-execution can be scheduled by invalidating a selected resource of the result of the previous execution. The invalidation can either be triggered by using schema coordinates for the root field (e.g. `Query.todos`) or by a resource identifier (e.g. `Todo:1`, todo with the id 1). **UPDATE:** [You can learn more how the library collects the resource identifiers here.](https://dev.to/n1ru4l/collecting-graphql-live-query-resource-identifier-with-graphql-tools-5fm5).

# LiveQueryStore Implementation

The `InMemoryLiveQueryStore` is a  class provided by the `@n1ru4l/in-memory-live-query-store` package `InMemoryLiveQueryStore` which stores all the information about the active live queries in memory.

## Registering the LiveQueryStore

In order to enable live queries for the `@n1ru4l/socket-io-graphql-server` implementation a `excute` function that returns a `AsyncIterator` must be passed to the `registerSocketIOGraphQLServer` function.

In addition to the default `execute` function from the `graphql-js` reference implementation, the new `execute` function can also return a `AsyncIterableIterator<ExecutionResult>`. **Note:** As `@defer` and `@stream` is added to the `graphql-js` reference implementation it now also can return `AsyncIterators`.

The `InMemoryLiveQueryStore` has the `execute` property which must be used for executing the live queries (but also non-live query operations by falling back to the default `execute`  from `graphql-js` if the operation is not identified as a live query operation).

```tsx
import socketIO from "socket.io";
import { InMemoryLiveQueryStore } from "@n1ru4l/in-memory-live-query-store";
import { registerSocketIOGraphQLServer } from "@n1ru4l/socket-io-graphql-server";
import { schema } from "./schema";

const liveQueryStore = new InMemoryLiveQueryStore();
const server = socketIO(httpServer);

registerSocketIOGraphQLServer({
  socketServer,
  // getExecutionParameter is invoked for each incoming operation.
  // a different context or even schema based on the connection can be returned
  // and will be used for that operation.
  getExecutionParameter: () => ({
    execute: liveQueryStore.execute,
    graphQLExecutionParameter: {
      schema,
      contextValue: {
        liveQueryStore,
      },
    },
  }),
});
```

There is a [WIP pull request on the `graphql-js` reference implementation](https://github.com/graphql/graphql-js/pull/2319) for adding `AsyncIterableIterator<ExecutionResult>` as a valid return result from `execute` for `@defer` and `@stream` directives. As a side effect, this would also help live queries.

## Notifying the live query store for changes

In order to re-execute the registered live queries and stream the new results to the connected clients, we must invalidate the live query operations, that select the specific data.

For the given query:

```graphql
query messages @live {
  messages(first: 10) {
    id
    content
    author {
      id
      name
    }
  }
}
```

That would look like this:

```ts
// Mutation.createMessage resolver
const createMessage = async (root, args, context) => {
  await context.messageStore.createMessage({
    content: args.content,
    userId: context.viewer.id,
  });
  // notify liveQueryStore via query schema coordinate that all live queries that select Queries.messages must be re-executed and sent to the clients.
  context.liveQueryStore.invalidate("Query.messages");
};

const editMessage = async (root, args, context) => {
  await context.messageStore.updateMessage({
    messageId: args.messageId,
    content: args.content,
  });
  // notify liveQueryStore via resource identifier
  context.liveQueryStore.invalidate(`Message:${args.messageId}`);
}
```

**Note:** It might be possible to do some kind of abstraction here. Imagine the live query store living on your mesh entry point keeping track of the live queries and mutations occurring and then automatically triggering the invalidations based on mutations instead of hard coding it into resolvers. [Apollo recently build something similar but quite a bit different](https://github.com/apollographql/real-time-federation).

# Adding the `@live` directive to your GraphQL schema

`@n1ru4l/graphql-live-query` exports a directive that can easily be added to an existing schema. You must add it, otherwise, the GraphQL server will complain about unknown directives in your GraphQL operations.

```tsx
import { GraphQLLiveDirective } from "@n1ru4l/graphql-live-query";

export const schema = new gql.GraphQLSchema({
  query,
  mutation,
  subscription,
  directives: [GraphQLLiveDirective],
});
```

For people that are using a GraphQL SDL driven development flow, you must add the following to your type definitions.

```graphql
directive @live on QUERY
```

# Consuming live queries on the client

The `@n1ru4l/socket-io-graphql-client` package can be used to execute (live) `Query`, `Mutation` and `Subscription` operations on to the server setup by `@n1ru4l/socket-io-graphql-server`. It implements the underlying GraphQL over Socket.io protocol.

[I also created a PR for supporting live queries with graphql-transport-ws](https://github.com/enisdenjo/graphql-transport-ws/pull/28).

## Basic client creation

```ts
import io from "socket.io-client";
import { createSocketIOGraphQLClient } from "@n1ru4l/socket-io-graphql-client";

const socket = io();

const socketIOGraphQLClient = createSocketIOGraphQLClient(socket);
```

## Executing GraphQL operations

The `SocketIOGraphQLClient` provides an `execute` method that will return an Observable which can be used to subscribe to the response.

A simple query or mutation operation will only publish one value. However, a live query or subscription will publish multiple values, therefore a Promise is not the perfect data structure for this.

The observable returned is compatible with the [proposal spec](https://github.com/tc39/proposal-observable) and is easily consumable by libraries such as apollo-client and relay. But also by tools like GraphiQL.

```tsx
socketIOGraphQLClient.execute({
  operation: /* GraphQL */ `
    query messages {
      id
      content
    }
  `
}, {
  next: console.log,
  error: console.log,
  complete: console.log
});

socketIOGraphQLClient.execute({
  operation: /* GraphQL */ `
    query messages @live {
      id
      content
    }
  `
}, {
  next: console.log,
  error: console.log,
  complete: console.log
});

socketIOGraphQLClient.execute({
  operation: /* GraphQL */ `
    subscription onNewMessage {
      onNewMessage {
        id
        content
      }
    }
  `
}, {
  next: console.log,
  error: console.log,
  complete: console.log
});
```

## GraphiQL fetcher

It is possible to easily consume and display the results of the GraphQL API with our `SocketIOGraphQLClient`. We just need to pass a custom fetcher to the GraphiQL component.

```tsx

const fetcher = ({ query: operation, ...restGraphQLParams }) =>
  ({
    subscribe: (
      sinkOrNext,
      ...args
    ) => {
      const sink: Sink =
        typeof sinkOrNext === "function"
          ? { next: sinkOrNext, error: args[0], complete: args[1] }
          : sinkOrNext;

      const unsubscribe = socketIOGraphQLClient.execute(
        {
          operation,
          ...restGraphQLParams,
        },
        sink
      );

      return { unsubscribe };
    },
  });

const CustomGraphiQL = () => (
  <GraphiQL
    fetcher={({ query: operation, ...execRest }) =>
      socketIOGraphQLClient.execute({ operation, ...execRest })
    }
  />
);
```

## Implementation with Relay

Relay is a powerful library for managing the cache client-side. A relay environment (which holds information about the cache and how the data is fetched from the server) can easily be built around a `SocketIOGraphQLClient` instance.

```tsx
import { SocketIOGraphQLClient } from "@n1ru4l/socket-io-graphql-client";
import {
  Environment,
  Network,
  RecordSource,
  Store,
  Observable,
  GraphQLResponse,
  RequestParameters,
  Variables,
} from "relay-runtime";

export const createRelayEnvironment = (
  networkInterface: SocketIOGraphQLClient<GraphQLResponse, Error>
) => {
  const execute = (request: RequestParameters, variables: Variables) => {
    if (!request.text) throw new Error("Missing document.");
    const { text: operation, name } = request;

    return Observable.create<GraphQLResponse>((sink) =>
      networkInterface.execute(
        {
          operation,
          variables,
          operationName: name,
        },
        sink
      )
    );
  };

  const network = Network.create(execute, execute);
  const store = attachNotifyGarbageCollectionBehaviourToStore(
    new Store(new RecordSource())
  );

  return new Environment({
    network,
    store,
  });
};
```

Consuming live data then becomes straight forward:

```ts
const ChatApplicationMessagesQuery = graphql`
  query ChatApplication_MessagesQuery @live {
    messages(limit: 10) {
      id
      ...ChatApplication_message
    }
  }
`;

const ChatApplicationMessageRenderer = React.memo(
  ({ message }: { message: ChatApplication_message }) => {
    return (
      <div>
        <div>{message.author.name}</div>
        <div>{message.content}</div>
      </div>
    );
  }
);

const ChatApplicationMessage = createFragmentContainer(
  ChatApplicationMessageRenderer,
  {
    message: graphql`
      fragment ChatApplication_message on Message {
        id
        content
        author {
          id
          name
        }
      }
    `,
  }
);

export const ChatApplication: React.FunctionComponent<{
  relayEnvironment: RelayEnvironment;
}> = (props) => {
  return (
    <QueryRenderer<ChatApplication_MessagesQuery>
      environment={props.relayEnvironment}
      query={ChatApplicationMessagesQuery}
      variables={{}}
      render={({ props }) => {
        if (!props) {
          return null;
        }

        return props.messages.map((message) => (
          <ChatApplicationMessage key={message.id} message={message} />
        ));
      }}
    />
  );
};
```

[See complete example app](https://github.com/n1ru4l/graphql-live-queries/tree/main/packages/todo-example/client-relay)

# Usage with Apollo Client

```ts
import { SocketIOGraphQLClient } from "@n1ru4l/socket-io-graphql-client";
import {
  ApolloClient,
  InMemoryCache,
  ApolloLink,
  Operation,
  Observable,
  FetchResult,
  Observable,
} from "@apollo/client";
import { print } from "graphql";

class SocketIOGraphQLApolloLink extends ApolloLink {
  private networkLayer: SocketIOGraphQLClient;
  constructor(networkLayer: SocketIOGraphQLClient) {
    super();
    this.networkLayer = networkLayer;
  }

  public request(operation: Operation): Observable<FetchResult> | null {
    return new Observable((sink) =>
      this.networkLayer.execute({
        operationName: operation.operationName,
        operation: print(operation.query),
        variables: operation.variables,
      })
    );
  }
}

export const createApolloClient = (networkInterface: SocketIOGraphQLClient) => {
  return new ApolloClient({
    link: new SocketIOGraphQLApolloLink(networkInterface),
    cache: new InMemoryCache(),
  });
};
```

[See complete example app](https://github.com/n1ru4l/graphql-live-queries/tree/main/packages/todo-example/client-apollo)

# Usage with Urql

```ts
import { SocketIOGraphQLClient } from "@n1ru4l/socket-io-graphql-client";
import {
  Client,
  dedupExchange,
  cacheExchange,
  subscriptionExchange,
  ExecutionResult,
} from "urql";

export const createUrqlClient = (
  networkInterface: SocketIOGraphQLClient<ExecutionResult>
) => {
  return new Client({
    url: "noop",
    exchanges: [
      dedupExchange,
      cacheExchange,
      subscriptionExchange({
        forwardSubscription: (operation) => ({
          subscribe: (sink) => ({
            unsubscribe: networkInterface.execute(
              {
                operation: operation.query,
                variables: operation.variables,
              },
              sink
            ),
          }),
        }),
        enableAllOperations: true,
      }),
    ],
  });
};
```

[See complete example app](https://github.com/n1ru4l/graphql-live-queries/tree/main/packages/todo-example/client-urql)

# The Future

This is the first implementation of the live query library. As more people try it out and build projects with it the API can become more mature and flexible for different use-cases.

New `LiveQueryStore` implementations for distributed systems could be built (e.g. based on Redis PubSub).

The network layer could furthermore be optimized to [only transport the updated patch instructions to the client in order to decrease the payload size](https://github.com/n1ru4l/graphql-live-queries/issues/7).

Other people might not use Socket.io. (Hybrid) implementations for other server libraries out there e.g. (apollo-server or express-graphql) must be built.

**Note:** GraphQL live queries can now be executed with [`graphql-ws`](https://github.com/enisdenjo/graphql-ws) and even [`express-graphql`](https://github.com/graphql/express-graphql/pull/583) **experimental**. Check out [GraphQL Bleeding Edge Playground](https://github.com/n1ru4l/graphql-bleeding-edge-playground)!

Do you have an idea? Feel free to contact me per [Twitter](https://twitter.com/n1rual), open a [GitHub issues](https://github.com/n1ru4l/graphql-live-queries) or write a comment down below 😊.