Chapter 7: Subscriptions

GraphQL subscriptions enable real-time data updates in your application. In this chapter, we'll explore how to implement and use subscriptions with GraphQL Yoga.

WebSocket Setup

Basic Setup

First, let's set up WebSocket support in our GraphQL server:

import { createYoga } from '@graphql-yoga/node'
import { createServer } from 'node:http'
import { WebSocketServer } from 'ws'

const yoga = createYoga({
  schema,
  plugins: [
    useGraphQLSSE() // Enable Server-Sent Events
  ]
})

const server = createServer(yoga)
const wss = new WebSocketServer({
  server,
  path: '/graphql'
})

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

Client Setup

On the client side, you can use the graphql-ws client:

import { createClient } from 'graphql-ws'
import WebSocket from 'ws'

const client = createClient({
  url: 'ws://localhost:4000/graphql',
  webSocketImpl: WebSocket
})

Subscription Types

Defining Subscription Types

Add subscription types to your schema:

type Subscription {
  postCreated: Post!
  postUpdated: Post!
  postDeleted: ID!
  commentAdded(postId: ID!): Comment!
}

type Post {
  id: ID!
  title: String!
  content: String!
  author: User!
  comments: [Comment!]!
}

type Comment {
  id: ID!
  content: String!
  author: User!
  post: Post!
}

Implementing Subscription Resolvers

import { PubSub } from 'graphql-subscriptions'

const pubsub = new PubSub()

const resolvers = {
  Subscription: {
    postCreated: {
      subscribe: () => pubsub.asyncIterator(['POST_CREATED'])
    },
    postUpdated: {
      subscribe: () => pubsub.asyncIterator(['POST_UPDATED'])
    },
    postDeleted: {
      subscribe: () => pubsub.asyncIterator(['POST_DELETED'])
    },
    commentAdded: {
      subscribe: (_, { postId }) => 
        pubsub.asyncIterator([`COMMENT_ADDED_${postId}`])
    }
  },
  
  Mutation: {
    createPost: async (_, { input }, context) => {
      const post = await context.prisma.post.create({
        data: input
      })
      
      pubsub.publish('POST_CREATED', {
        postCreated: post
      })
      
      return post
    },
    
    addComment: async (_, { postId, content }, context) => {
      const comment = await context.prisma.comment.create({
        data: {
          content,
          postId,
          authorId: context.user.id
        }
      })
      
      pubsub.publish(`COMMENT_ADDED_${postId}`, {
        commentAdded: comment
      })
      
      return comment
    }
  }
}

Advanced Subscription Patterns

Filtered Subscriptions

Filter subscriptions based on user permissions:

const resolvers = {
  Subscription: {
    postCreated: {
      subscribe: withFilter(
        () => pubsub.asyncIterator(['POST_CREATED']),
        (payload, _, context) => {
          // Only subscribe to posts from followed users
          return context.user.following.includes(payload.postCreated.authorId)
        }
      )
    }
  }
}

Subscription with Arguments

Handle subscription arguments:

const resolvers = {
  Subscription: {
    postUpdated: {
      subscribe: withFilter(
        () => pubsub.asyncIterator(['POST_UPDATED']),
        (payload, { postId }, context) => {
          // Only subscribe to specific post updates
          return payload.postUpdated.id === postId
        }
      )
    }
  }
}

Real-time Features

Live Comments

Implement live comments with subscriptions:

// Schema
const typeDefs = `
  type Subscription {
    commentAdded(postId: ID!): Comment!
  }
  
  type Comment {
    id: ID!
    content: String!
    author: User!
    post: Post!
    createdAt: DateTime!
  }
`

// Resolver
const resolvers = {
  Subscription: {
    commentAdded: {
      subscribe: withFilter(
        (_, { postId }) => pubsub.asyncIterator([`COMMENT_ADDED_${postId}`]),
        (payload, { postId }) => payload.commentAdded.postId === postId
      )
    }
  }
}

// Client usage
const COMMENTS_SUBSCRIPTION = gql`
  subscription OnCommentAdded($postId: ID!) {
    commentAdded(postId: $postId) {
      id
      content
      author {
        id
        name
      }
      createdAt
    }
  }
`

function PostComments({ postId }) {
  const { data } = useSubscription(COMMENTS_SUBSCRIPTION, {
    variables: { postId }
  })
  
  return (
    <div>
      {data?.commentAdded && (
        <Comment comment={data.commentAdded} />
      )}
    </div>
  )
}

Live Notifications

Implement real-time notifications:

// Schema
const typeDefs = `
  type Subscription {
    notificationReceived: Notification!
  }
  
  type Notification {
    id: ID!
    type: NotificationType!
    message: String!
    read: Boolean!
    createdAt: DateTime!
  }
  
  enum NotificationType {
    LIKE
    COMMENT
    FOLLOW
    MENTION
  }
`

// Resolver
const resolvers = {
  Subscription: {
    notificationReceived: {
      subscribe: (_, __, context) => {
        if (!context.user) {
          throw new AuthenticationError('Not authenticated')
        }
        return pubsub.asyncIterator([`NOTIFICATION_${context.user.id}`])
      }
    }
  }
}

// Client usage
const NOTIFICATION_SUBSCRIPTION = gql`
  subscription OnNotificationReceived {
    notificationReceived {
      id
      type
      message
      read
      createdAt
    }
  }
`

function NotificationCenter() {
  const { data } = useSubscription(NOTIFICATION_SUBSCRIPTION)
  
  return (
    <div>
      {data?.notificationReceived && (
        <Notification notification={data.notificationReceived} />
      )}
    </div>
  )
}

Deployment Considerations

Serverless vs. Persistent Connections

When deploying subscriptions, consider:

Serverless Environments
- Use WebSocket-compatible serverless platforms
- Consider using a message broker (Redis, RabbitMQ)
- Implement connection pooling
Traditional Servers
- Handle WebSocket connections efficiently
- Implement proper error handling
- Monitor connection health

Scaling Subscriptions

For scaling subscriptions:

import { RedisPubSub } from 'graphql-redis-subscriptions'
import Redis from 'ioredis'

const options = {
  host: process.env.REDIS_HOST,
  port: process.env.REDIS_PORT,
  password: process.env.REDIS_PASSWORD
}

const pubsub = new RedisPubSub({
  publisher: new Redis(options),
  subscriber: new Redis(options)
})

const resolvers = {
  Subscription: {
    postCreated: {
      subscribe: () => pubsub.asyncIterator(['POST_CREATED'])
    }
  }
}

Connection Management

Handle WebSocket connections properly:

const wss = new WebSocketServer({
  server,
  path: '/graphql',
  // Handle connection lifecycle
  onConnect: (ctx) => {
    console.log('Client connected')
  },
  onDisconnect: (ctx) => {
    console.log('Client disconnected')
  }
})

// Implement heartbeat
setInterval(() => {
  wss.clients.forEach((client) => {
    if (client.isAlive === false) {
      return client.terminate()
    }
    client.isAlive = false
    client.ping()
  })
}, 30000)

Best Practices

Error Handling
- Handle connection errors
- Implement reconnection logic
- Log subscription errors
Performance
- Use connection pooling
- Implement proper cleanup
- Monitor subscription usage
Security
- Authenticate subscriptions
- Validate subscription arguments
- Rate limit subscriptions
Monitoring
- Track active subscriptions
- Monitor WebSocket connections
- Log subscription events

Example: Complete Subscription Setup

Here's a complete example of a subscription setup:

// server.ts
import { createYoga } from '@graphql-yoga/node'
import { createServer } from 'node:http'
import { WebSocketServer } from 'ws'
import { RedisPubSub } from 'graphql-redis-subscriptions'
import Redis from 'ioredis'

const redisOptions = {
  host: process.env.REDIS_HOST,
  port: process.env.REDIS_PORT
}

const pubsub = new RedisPubSub({
  publisher: new Redis(redisOptions),
  subscriber: new Redis(redisOptions)
})

const yoga = createYoga({
  schema,
  plugins: [
    useGraphQLSSE()
  ]
})

const server = createServer(yoga)
const wss = new WebSocketServer({
  server,
  path: '/graphql'
})

// schema.ts
const typeDefs = `
  type Subscription {
    postCreated: Post!
    postUpdated: Post!
    postDeleted: ID!
    commentAdded(postId: ID!): Comment!
  }
  
  type Post {
    id: ID!
    title: String!
    content: String!
    author: User!
    comments: [Comment!]!
  }
  
  type Comment {
    id: ID!
    content: String!
    author: User!
    post: Post!
  }
`

// resolvers.ts
const resolvers = {
  Subscription: {
    postCreated: {
      subscribe: withFilter(
        () => pubsub.asyncIterator(['POST_CREATED']),
        (payload, _, context) => {
          return context.user.following.includes(payload.postCreated.authorId)
        }
      )
    },
    postUpdated: {
      subscribe: withFilter(
        () => pubsub.asyncIterator(['POST_UPDATED']),
        (payload, { postId }) => {
          return payload.postUpdated.id === postId
        }
      )
    },
    commentAdded: {
      subscribe: withFilter(
        (_, { postId }) => pubsub.asyncIterator([`COMMENT_ADDED_${postId}`]),
        (payload, { postId }) => {
          return payload.commentAdded.postId === postId
        }
      )
    }
  },
  
  Mutation: {
    createPost: async (_, { input }, context) => {
      const post = await context.prisma.post.create({
        data: {
          ...input,
          authorId: context.user.id
        }
      })
      
      pubsub.publish('POST_CREATED', {
        postCreated: post
      })
      
      return post
    },
    
    addComment: async (_, { postId, content }, context) => {
      const comment = await context.prisma.comment.create({
        data: {
          content,
          postId,
          authorId: context.user.id
        }
      })
      
      pubsub.publish(`COMMENT_ADDED_${postId}`, {
        commentAdded: comment
      })
      
      return comment
    }
  }
}

Summary

In this chapter, we've covered:

WebSocket setup with GraphQL Yoga
Implementing subscription types and resolvers
Advanced subscription patterns
Real-time features implementation
Deployment considerations
Best practices for subscriptions

These patterns will help you build real-time features in your GraphQL applications. In the next chapter, we'll explore integrations with various frameworks and tools.

WebSocket Setup​

Basic Setup​

Client Setup​

Subscription Types​

Defining Subscription Types​

Implementing Subscription Resolvers​

Advanced Subscription Patterns​

Filtered Subscriptions​

Subscription with Arguments​

Real-time Features​

Live Comments​

Live Notifications​

Deployment Considerations​

Serverless vs. Persistent Connections​

Scaling Subscriptions​

Connection Management​

Best Practices​

Example: Complete Subscription Setup​

Summary​