GraphQL Subscriptions

type Subscription {
  # Subscribe to new messages in a chat room
  messageAdded(roomId: ID!): Message!

  # Subscribe to post updates (created, updated, deleted)
  postChanged: PostChangeEvent!

  # Subscribe to notifications for the authenticated user
  notificationReceived: Notification!

  # Subscribe to typing indicators
  userTyping(roomId: ID!): TypingEvent!
}

type Message {
  id: ID!
  text: String!
  sender: User!
  room: ChatRoom!
  createdAt: String!
}

type PostChangeEvent {
  type: ChangeType!
  post: Post!
}

enum ChangeType {
  CREATED
  UPDATED
  DELETED
}

type Notification {
  id: ID!
  type: NotificationType!
  message: String!
  link: String
  read: Boolean!
  createdAt: String!
}

type TypingEvent {
  user: User!
  isTyping: Boolean!
}

import { createServer } from 'http';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { ApolloServer } from '@apollo/server';
import { expressMiddleware } from '@apollo/server/express4';
import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer';
import { makeExecutableSchema } from '@graphql-tools/schema';
import { PubSub, withFilter } from 'graphql-subscriptions';
import express from 'express';

// Create a PubSub instance for event publishing
const pubsub = new PubSub();

// Event name constants
const EVENTS = {
  MESSAGE_ADDED: 'MESSAGE_ADDED',
  POST_CHANGED: 'POST_CHANGED',
  NOTIFICATION: 'NOTIFICATION',
  USER_TYPING: 'USER_TYPING',
} as const;

const schema = makeExecutableSchema({ typeDefs, resolvers });

// Set up Express and HTTP server
const app = express();
const httpServer = createServer(app);

// Set up WebSocket server for subscriptions
const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/graphql',
});

const serverCleanup = useServer(
  {
    schema,
    context: async (ctx) => {
      // Authenticate WebSocket connections
      const token = ctx.connectionParams?.authToken as string;
      const currentUser = token ? await verifyToken(token) : null;
      return { currentUser, pubsub };
    },
  },
  wsServer
);

// Set up Apollo Server
const server = new ApolloServer({
  schema,
  plugins: [
    ApolloServerPluginDrainHttpServer({ httpServer }),
    {
      async serverWillStart() {
        return {
          async drainServer() {
            await serverCleanup.dispose();
          },
        };
      },
    },
  ],
});

await server.start();
app.use('/graphql', expressMiddleware(server));
httpServer.listen(4000);

const resolvers = {
  Subscription: {
    // Simple subscription — all events go to all subscribers
    postChanged: {
      subscribe: () => pubsub.asyncIterableIterator([EVENTS.POST_CHANGED]),
    },

    // Filtered subscription — only events for a specific room
    messageAdded: {
      subscribe: withFilter(
        () => pubsub.asyncIterableIterator([EVENTS.MESSAGE_ADDED]),
        (payload, variables) => {
          // Only send to subscribers watching this specific room
          return payload.messageAdded.roomId === variables.roomId;
        }
      ),
    },

    // Authenticated subscription — only events for the current user
    notificationReceived: {
      subscribe: withFilter(
        () => pubsub.asyncIterableIterator([EVENTS.NOTIFICATION]),
        (payload, _variables, context) => {
          // Only send notifications to the intended recipient
          return payload.notificationReceived.userId === context.currentUser?.id;
        }
      ),
    },

    // Typing indicator with room filtering
    userTyping: {
      subscribe: withFilter(
        () => pubsub.asyncIterableIterator([EVENTS.USER_TYPING]),
        (payload, variables, context) => {
          return (
            payload.userTyping.roomId === variables.roomId &&
            payload.userTyping.user.id !== context.currentUser?.id
          );
        }
      ),
    },
  },

  // Mutations that publish events
  Mutation: {
    sendMessage: async (_: unknown, args: { roomId: string; text: string }, ctx: Context) => {
      const message = await ctx.prisma.message.create({
        data: {
          text: args.text,
          roomId: args.roomId,
          senderId: ctx.currentUser!.id,
        },
        include: { sender: true, room: true },
      });

      // Publish event to all subscribers of this room
      pubsub.publish(EVENTS.MESSAGE_ADDED, {
        messageAdded: { ...message, roomId: args.roomId },
      });

      return message;
    },

    createPost: async (_: unknown, args: { input: CreatePostInput }, ctx: Context) => {
      const post = await ctx.prisma.post.create({
        data: { ...args.input, authorId: ctx.currentUser!.id },
      });

      pubsub.publish(EVENTS.POST_CHANGED, {
        postChanged: { type: 'CREATED', post },
      });

      return post;
    },
  },
};

import { useSubscription, gql } from '@apollo/client';
import { GraphQLWsLink } from '@apollo/client/link/subscriptions';
import { createClient } from 'graphql-ws';
import { split, HttpLink } from '@apollo/client';
import { getMainDefinition } from '@apollo/client/utilities';

// Set up WebSocket link for subscriptions
const wsLink = new GraphQLWsLink(
  createClient({
    url: 'ws://localhost:4000/graphql',
    connectionParams: {
      authToken: localStorage.getItem('token'),
    },
  })
);

const httpLink = new HttpLink({ uri: 'http://localhost:4000/graphql' });

// Split traffic: subscriptions go over WebSocket, everything else over HTTP
const splitLink = split(
  ({ query }) => {
    const definition = getMainDefinition(query);
    return (
      definition.kind === 'OperationDefinition' &&
      definition.operation === 'subscription'
    );
  },
  wsLink,
  httpLink,
);

// React component using subscriptions
const MESSAGE_SUBSCRIPTION = gql`
  subscription OnMessageAdded($roomId: ID!) {
    messageAdded(roomId: $roomId) {
      id
      text
      sender { id name avatarUrl }
      createdAt
    }
  }
`;

function ChatRoom({ roomId }: { roomId: string }) {
  const { data, loading, error } = useSubscription(MESSAGE_SUBSCRIPTION, {
    variables: { roomId },
  });

  if (loading) return <p>Connecting to chat...</p>;
  if (error) return <p>Connection error: {error.message}</p>;

  const message = data?.messageAdded;

  return (
    <div>
      {message && (
        <div className="flex gap-3 p-3">
          <img src={message.sender.avatarUrl} className="w-8 h-8 rounded-full" />
          <div>
            <span className="font-bold">{message.sender.name}</span>
            <p>{message.text}</p>
          </div>
        </div>
      )}
    </div>
  );
}