Subscriptions

# 1) Schema
type Subscription {
    postCreated:   Post!
    commentAdded(postId: ID!): Comment!
    presenceChanged(roomId: ID!): Presence!
}

# 2) Apollo Server — WebSocket setup
import { ApolloServer } from '@apollo/server';
import { expressMiddleware } from '@apollo/server/express4';
import { makeExecutableSchema } from '@graphql-tools/schema';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { PubSub } from 'graphql-subscriptions';
import http from 'node:http';
import express from 'express';

const pubsub = new PubSub();
const PUB = {
    POST_CREATED:   'POST_CREATED',
    COMMENT_ADDED:  'COMMENT_ADDED:',          // suffix with postId
};

const resolvers = {
    Query:    { /* … */ },
    Mutation: {
        async createPost(_, { input }, ctx) {
            const post = await db.posts.create(input);
            pubsub.publish(PUB.POST_CREATED, { postCreated: post });
            return post;
        },
        async addComment(_, { postId, body }, ctx) {
            const c = await db.comments.create({ postId, body, authorId: ctx.user.id });
            pubsub.publish(`${PUB.COMMENT_ADDED}${postId}`, { commentAdded: c });
            return c;
        },
    },
    Subscription: {
        postCreated: {
            subscribe: () => pubsub.asyncIterableIterator([PUB.POST_CREATED]),
        },
        commentAdded: {
            subscribe: (_, { postId }) =>
                pubsub.asyncIterableIterator([`${PUB.COMMENT_ADDED}${postId}`]),
        },
    },
};

const schema = makeExecutableSchema({ typeDefs, resolvers });
const app    = express();
const http_  = http.createServer(app);

# WebSocket endpoint
const wsServer = new WebSocketServer({ server: http_, path: '/graphql' });
const serverCleanup = useServer({
    schema,
    context: async (ctx, _msg, _args) => {
        const user = await verifyToken(ctx.connectionParams?.token);
        return { user, db };
    },
    onConnect: async (ctx) => {
        if (!await verifyToken(ctx.connectionParams?.token)) return false;   // refuse
    },
}, wsServer);

const apollo = new ApolloServer({
    schema,
    plugins: [{
        async serverWillStart() {
            return { async drainServer() { await serverCleanup.dispose(); } };
        },
    }],
});
await apollo.start();
app.use('/graphql', express.json(), expressMiddleware(apollo, {
    context: async ({ req }) => ({ user: await verifyToken(req.headers.authorization), db }),
}));
http_.listen(4000);

# 3) Run a subscription (graphql-ws over WS)
subscription OnComment($postId: ID!) {
    commentAdded(postId: $postId) {
        id
        body
        createdAt
        author { name }
    }
}

# 4) Client — Apollo
import { ApolloClient, InMemoryCache, split, HttpLink } from '@apollo/client';
import { getMainDefinition } from '@apollo/client/utilities';
import { GraphQLWsLink }    from '@apollo/client/link/subscriptions';
import { createClient }     from 'graphql-ws';

const wsLink = new GraphQLWsLink(createClient({
    url: 'wss://example.com/graphql',
    connectionParams: () => ({ token: getAuthToken() }),
    retryAttempts: Infinity,
}));
const httpLink = new HttpLink({ uri: 'https://example.com/graphql' });

const splitLink = split(
    ({ query }) => {
        const def = getMainDefinition(query);
        return def.kind === 'OperationDefinition' && def.operation === 'subscription';
    },
    wsLink,
    httpLink,
);

export const client = new ApolloClient({ link: splitLink, cache: new InMemoryCache() });

// In a React component
const { data } = useSubscription(ON_COMMENT, { variables: { postId } });

# 5) Server-Sent Events (SSE) — simpler than WS for one-way updates
# graphql-sse package supports SSE; useful when WS infrastructure is missing.

# 6) Scaling — beyond a single process
# In-memory PubSub only works on ONE node. Multi-process:
#   - Redis PubSub (graphql-redis-subscriptions)
#   - Kafka / NATS / RabbitMQ for durability
#   - Use sticky sessions on the LB; or stateless WS + bus

# 7) Auth + authorisation
#   - Authenticate at connection time (connectionParams.token)
#   - Re-check authz per subscription (e.g. user must own the chat room)
#   - Filter events in the resolver if not all subscribers should see all events

# 8) When NOT to use subscriptions
#   • Refreshing data every few seconds — polling is simpler
#   • One-off notifications — push notifications via FCM / APNs / web push
#   • Massive fan-out — consider WebSocket + custom protocol or SSE

# 9) Best practices
#   • Always rate-limit messages per connection
#   • Send heartbeats (keep-alive every 30s)
#   • Reconnect with exponential backoff on the client
#   • Test with WS disconnect → reconnect mid-stream

subscription OnNewMessage {
    newMessage { id body author { name } }
}