GraphQL Subscriptions in Production: Real-Time APIs with Redis Pub/Sub and Pothos

GraphQL subscriptions are the right tool for a narrow set of problems: live dashboard updates, collaborative editing presence, order status tracking, and chat. For most data-fetching needs, queries and mutations are simpler and more reliable. But when you genuinely need server-to-client push, subscriptions deliver it within the GraphQL contract your frontend already understands.

The challenge is scaling. A single server maintains WebSocket connections in memory — fine for one instance, broken when you add a second. Redis pub/sub bridges the gap: any server can publish an event, and all servers with subscribers for that event receive and forward it.

This post covers a production-ready setup with GraphQL Yoga, Pothos type-safe schema builder, and Redis-backed pub/sub that scales horizontally.

---

When to Use GraphQL Subscriptions (vs. Alternatives)

If your app already uses GraphQL, subscriptions are the natural fit. If it doesn't, SSE is simpler for server-push.

---

Stack

• GraphQL Yoga — production-ready GraphQL server built on Fetch API (works with Fastify, Node.js, Cloudflare Workers)
• Pothos — code-first, type-safe GraphQL schema builder (no code generation needed)
• graphql-ws — WebSocket subprotocol (graphql-transport-ws) — the 2026 standard
• Redis — pub/sub backbone for multi-instance scaling
• ioredis — Redis client with reliable reconnection

---

Schema Definition with Pothos

// src/schema/builder.ts
import SchemaBuilder from '@pothos/core';
import PrismaPlugin from '@pothos/plugin-prisma';
import type PrismaTypes from '@pothos/plugin-prisma/generated';
import { db } from '@/db/prisma';

export const builder = new SchemaBuilder<{
  PrismaTypes: PrismaTypes;
  Context: {
    userId: string;
    tenantId: string;
  };
  Scalars: {
    DateTime: { Input: Date; Output: Date };
    JSON: { Input: unknown; Output: unknown };
  };
}>({
  plugins: [PrismaPlugin],
  prisma: { client: db },
});

builder.queryType({});
builder.mutationType({});
builder.subscriptionType({});

// src/schema/order/types.ts
import { builder } from '../builder';

export const OrderStatus = builder.enumType('OrderStatus', {
  values: ['PENDING', 'CONFIRMED', 'SHIPPED', 'DELIVERED', 'CANCELLED'] as const,
});

export const OrderType = builder.prismaObject('Order', {
  fields: (t) => ({
    id: t.exposeID('id'),
    status: t.expose('status', { type: OrderStatus }),
    total: t.exposeFloat('total'),
    createdAt: t.expose('createdAt', { type: 'DateTime' }),
    customer: t.relation('customer'),
    items: t.relation('items'),
  }),
});

export const OrderEventType = builder.objectType('OrderEvent', {
  fields: (t) => ({
    orderId: t.exposeString('orderId'),
    status: t.expose('status', { type: OrderStatus }),
    timestamp: t.expose('timestamp', { type: 'DateTime' }),
    message: t.exposeString('message', { nullable: true }),
  }),
});

---

Redis Pub/Sub Layer

// src/pubsub/redis-pubsub.ts
import Redis from 'ioredis';
import { EventEmitter } from 'events';

type EventPayload = Record<string, unknown>;

export class RedisPubSub {
  private publisher: Redis;
  private subscriber: Redis;
  private emitter = new EventEmitter();

  constructor(redisUrl: string) {
    const opts = {
      maxRetriesPerRequest: null,
      enableReadyCheck: false,
      retryStrategy: (times: number) => Math.min(times * 100, 3000),
    };

    this.publisher = new Redis(redisUrl, opts);
    this.subscriber = new Redis(redisUrl, opts);

    // Forward Redis messages to local EventEmitter
    this.subscriber.on('message', (channel, message) => {
      try {
        const payload = JSON.parse(message);
        this.emitter.emit(channel, payload);
      } catch {
        // Ignore malformed messages
      }
    });
  }

  async publish(channel: string, payload: EventPayload): Promise<void> {
    await this.publisher.publish(channel, JSON.stringify(payload));
  }

  async subscribe(channel: string): Promise<void> {
    await this.subscriber.subscribe(channel);
  }

  async unsubscribe(channel: string): Promise<void> {
    await this.subscriber.unsubscribe(channel);
  }

  on(channel: string, listener: (payload: EventPayload) => void): void {
    this.emitter.on(channel, listener);
  }

  off(channel: string, listener: (payload: EventPayload) => void): void {
    this.emitter.off(channel, listener);
  }

  // Create an async generator for use in GraphQL subscriptions
  async *asyncIterator<T extends EventPayload>(
    channel: string,
  ): AsyncGenerator<T> {
    await this.subscribe(channel);

    const queue: T[] = [];
    let resolve: (() => void) | null = null;
    let done = false;

    const listener = (payload: T) => {
      queue.push(payload);
      resolve?.();
      resolve = null;
    };

    this.on(channel, listener as (p: EventPayload) => void);

    try {
      while (!done) {
        if (queue.length > 0) {
          yield queue.shift()!;
        } else {
          await new Promise<void>((r) => { resolve = r; });
        }
      }
    } finally {
      this.off(channel, listener as (p: EventPayload) => void);
      await this.unsubscribe(channel);
    }
  }
}

export const pubsub = new RedisPubSub(process.env.REDIS_URL!);

---

Subscription Resolvers

// src/schema/order/subscriptions.ts
import { builder } from '../builder';
import { pubsub } from '@/pubsub/redis-pubsub';
import { OrderEventType } from './types';
import { verifyOrderAccess } from '@/services/authorization';

interface OrderEvent {
  orderId: string;
  status: string;
  timestamp: Date;
  message?: string;
}

builder.subscriptionField('orderUpdated', (t) =>
  t.field({
    type: OrderEventType,
    args: {
      orderId: t.arg.string({ required: true }),
    },
    // Auth check before subscribing
    async authScopes(_, { orderId }, { userId }) {
      return verifyOrderAccess(userId, orderId);
    },
    subscribe: async function* (_, { orderId }, { userId }) {
      const channel = `order:${orderId}:events`;

      // Yield current order status immediately (no missed events)
      const order = await db.order.findUnique({ where: { id: orderId } });
      if (order) {
        yield {
          orderId: order.id,
          status: order.status,
          timestamp: new Date(),
          message: 'Current status',
        } satisfies OrderEvent;
      }

      // Then stream future events from Redis
      yield* pubsub.asyncIterator<OrderEvent>(channel);
    },
    resolve: (event) => event,
  }),
);

// Subscribe to ALL orders for a tenant (admin dashboard)
builder.subscriptionField('tenantOrderFeed', (t) =>
  t.field({
    type: OrderEventType,
    authScopes: { isAdmin: true },
    subscribe: async function* (_, __, { tenantId }) {
      yield* pubsub.asyncIterator<OrderEvent>(`tenant:${tenantId}:orders`);
    },
    resolve: (event) => event,
  }),
);

---

Publishing Events from Business Logic

// src/services/orders.ts
import { pubsub } from '@/pubsub/redis-pubsub';
import { db } from '@/db/prisma';

export async function shipOrder(
  orderId: string,
  trackingNumber: string,
): Promise<void> {
  const order = await db.order.update({
    where: { id: orderId },
    data: { status: 'SHIPPED', trackingNumber },
    include: { customer: true },
  });

  const event = {
    orderId,
    status: 'SHIPPED',
    timestamp: new Date(),
    message: `Your order has shipped. Tracking: ${trackingNumber}`,
  };

  // Publish to order-specific channel (watched by customer)
  await pubsub.publish(`order:${orderId}:events`, event);

  // Also publish to tenant feed (watched by admin dashboard)
  await pubsub.publish(`tenant:${order.tenantId}:orders`, event);
}

---

GraphQL Yoga Server with WebSocket Support

// src/server.ts
import Fastify from 'fastify';
import { createYoga } from 'graphql-yoga';
import { useServer } from 'graphql-ws/lib/use/fastify-websocket';
import fastifyWebsocket from '@fastify/websocket';
import { schema } from './schema';
import { createContext } from './context';

const app = Fastify({ logger: true });
await app.register(fastifyWebsocket);

const yoga = createYoga({
  schema,
  context: createContext,
  graphqlEndpoint: '/graphql',
  subscriptionsProtocol: 'WS',
  logging: app.log,
});

// HTTP + WebSocket on same endpoint
app.route({
  url: yoga.graphqlEndpoint,
  method: ['GET', 'POST', 'OPTIONS'],
  handler: async (req, reply) => {
    const response = await yoga.handleNodeRequestAndResponse(req, reply, {
      req,
      reply,
    });
    return response;
  },
  wsHandler: useServer(
    {
      schema,
      context: async (ctx) => {
        // Authenticate WebSocket connection via connectionParams
        const token = ctx.connectionParams?.authorization as string;
        if (!token) throw new Error('Unauthorized');

        const user = await verifyJWT(token);
        return { userId: user.sub, tenantId: user.tenantId };
      },
      onConnect: (ctx) => {
        app.log.info({ ip: ctx.extra.socket.remoteAddress }, 'WS connected');
      },
      onDisconnect: (ctx) => {
        app.log.info({ ip: ctx.extra.socket.remoteAddress }, 'WS disconnected');
      },
    },
    app.websocketServer,
  ),
});

await app.listen({ port: 4000, host: '0.0.0.0' });

---

React Client with urql

// src/providers/graphql.tsx
import { createClient, subscriptionExchange, cacheExchange, fetchExchange } from 'urql';
import { createClient as createWSClient } from 'graphql-ws';

const wsClient = createWSClient({
  url: process.env.NEXT_PUBLIC_WS_URL ?? 'wss://api.myapp.com/graphql',
  connectionParams: async () => ({
    // JWT from auth store
    authorization: await getAuthToken(),
  }),
  retryAttempts: 5,
  on: {
    error: (err) => console.error('WS error:', err),
  },
});

export const gqlClient = createClient({
  url: process.env.NEXT_PUBLIC_API_URL ?? 'https://api.myapp.com/graphql',
  exchanges: [
    cacheExchange,
    fetchExchange,
    subscriptionExchange({
      forwardSubscription: (request) => ({
        subscribe: (sink) => ({
          unsubscribe: wsClient.subscribe(request as any, sink),
        }),
      }),
    }),
  ],
});

// src/components/OrderTracker.tsx
import { useSubscription } from 'urql';
import { gql } from 'urql';

const ORDER_UPDATED = gql`
  subscription OrderUpdated($orderId: String!) {
    orderUpdated(orderId: $orderId) {
      orderId
      status
      timestamp
      message
    }
  }
`;

export function OrderTracker({ orderId }: { orderId: string }) {
  const [result] = useSubscription({
    query: ORDER_UPDATED,
    variables: { orderId },
  });

  const event = result.data?.orderUpdated;

  return (
    <div className="order-tracker">
      {result.fetching && <p>Connecting...</p>}
      {result.error && <p className="text-red-600">Connection error</p>}
      {event && (
        <div>
          <StatusBadge status={event.status} />
          <p className="text-sm text-gray-500">
            {new Date(event.timestamp).toLocaleTimeString()}
          </p>
          {event.message && <p>{event.message}</p>}
        </div>
      )}
    </div>
  );
}

---

Scaling: Connection Counts and Infrastructure

WebSocket connections are stateful — each connection stays open. At 10,000 concurrent connections per instance (Node.js limit varies by memory/CPU), plan capacity accordingly.

Load balancer config for sticky sessions (AWS ALB):

resource "aws_alb_target_group" "graphql_ws" {
  name             = "graphql-ws"
  port             = 4000
  protocol         = "HTTP"
  target_type      = "ip"

  stickiness {
    type            = "lb_cookie"
    cookie_duration = 86400   # 24h — WebSocket reconnects go to same instance
    enabled         = true
  }

  health_check {
    path                = "/health"
    healthy_threshold   = 2
    unhealthy_threshold = 3
    interval            = 10
  }
}

---

Working With Viprasol

Our team builds real-time GraphQL APIs for collaborative SaaS products — from live dashboards to order tracking to multiplayer features — with Redis-backed scaling and production-tested WebSocket infrastructure.

What we deliver:

• GraphQL Yoga + Pothos type-safe schema with subscriptions
• Redis pub/sub layer with reliable reconnection
• Authentication for WebSocket connections (JWT via connectionParams)
• Horizontal scaling setup with sticky sessions
• React/Next.js client integration with urql or Apollo

→ Discuss your real-time API requirements → Web development services

---

See Also

• WebSockets vs Server-Sent Events
• GraphQL Federation
• Real-Time Analytics with ClickHouse and Kafka
• API Gateway Patterns
• Redis Caching Strategies