Real-Time GraphQL Subscriptions & Federation

+-----------------------------------------------------------------------------------+
| GraphQL Subscriptions Network Architecture                                        |
|                                                                                   |
|  [ Client Application: React / Apollo Client ]                                    |
|                         |                                                         |
|                         | 1. Persistent Bi-directional WebSocket / SSE Handshake |
|                         |    Protocol: 'graphql-transport-ws' (RFC compliant)     |
|                         v                                                         |
|  +-----------------------------------------------------------------------------+  |
|  | GraphQL Subscription Gateway / WebSocket Server                             |  |
|  | - Handles connection initialization (connection_init -> connection_ack)     |  |
|  | - Maintains active client socket registry (Socket ID -> Subscriptions Map)  |  |
|  | - Dispatches heartbeat PING/PONG frames to detect stale half-open sockets   |  |
|  +-------------------------------------+---------------------------------------+  |
|                                        |                                          |
|                                        | 2. Subscribes to event topics            |
|                                        v                                          |
|  +-----------------------------------------------------------------------------+  |
|  | Distributed Event Bus: Redis Pub/Sub / Kafka / RabbitMQ                     |  |
|  | - Topic: "CHAT_MESSAGE_CREATED:channel_99"                                   |  |
|  +-------------------------------------+---------------------------------------+  |
|                                        ^                                          |
|                                        | 3. Mutation publishes new event          |
|  +-------------------------------------+---------------------------------------+  |
|  | Mutating Backend Server (HTTP POST /graphql)                                |  |
|  | - mutation { postMessage(...) } -> Saves to DB -> pubsub.publish(...)       |  |
|  +-----------------------------------------------------------------------------+  |
|                                                                                   |
|  Data Flow (Steps 3 -> 2 -> 1):                                                   |
|  Mutation publishes payload to Redis -> Redis distributes to all Gateway nodes -> |
|  Gateway evaluates subscription filter -> Pushes serialized GraphQL JSON to client|
+-----------------------------------------------------------------------------------+

const resolvers = {
  Subscription: {
    messageAdded: {
      // 1. Subscribes to the underlying event stream (AsyncIterator)
      subscribe: (_parent, args, context) => {
        return context.pubsub.asyncIterator([`CHANNEL_{args.channelId}`]);
      },
      // 2. Maps the raw event payload into the GraphQL response shape
      resolve: (payload) => {
        return payload.messageCreated;
      }
    }
  }
};

  [ Client A (WebSocket) ]                  [ Client B (HTTP Client) ]
             |                                          |
             v                                          v
      +--------------+                           +--------------+
      | Pod 1 (Node) |                           | Pod 2 (Node) |
      | Socket A     |                           | Mutation     |
      +-------+------+                           +-------+------+
              |                                          |
              |                                          | In-Memory PubSub
              |                                          | Emits to Pod 2 ONLY!
              x <========================================+
                 Pod 1 NEVER receives the event!
                 Client A NEVER receives the update!

+-----------------------------------------------------------------------------------+
| Monolithic GraphQL vs Apollo Federation v2 Architecture                           |
|                                                                                   |
|  Monolithic Schema Repository:                                                    |
|  - 1 Huge Server (50,000 lines of SDL)                                            |
|  - 15 Teams committing to 1 repo                                                  |
|  - If Team A introduces a bug, Team B's service deploy is blocked                 |
|                                                                                   |
|  -------------------------------------------------------------------------------  |
|                                                                                   |
|  Apollo Federation v2 Supergraph Architecture:                                    |
|                                                                                   |
|                             [ Client Query Request ]                              |
|                                        |                                          |
|                                        v                                          |
|                     +--------------------------------------+                      |
|                     | Apollo Router / Gateway (Rust-based) |                      |
|                     | - Exposes unified Supergraph Schema  |                      |
|                     | - Query Planner: Generates DAG plan  |                      |
|                     | - Orchestrates downstream subgraphs  |                      |
|                     +------------------+-------------------+                      |
|                                        |                                          |
|             +--------------------------+--------------------------+               |
|             | HTTP POST (Sub-query)    | HTTP POST (Sub-query)    | HTTP POST     |
|             v                          v                          v               |
|    +------------------+       +------------------+       +------------------+     |
|    | Users Subgraph   |       | Products Subgraph|       | Reviews Subgraph |     |
|    | (Auth Team)      |       | (Inventory Team) |       | (Community Team) |     |
|    | Owns: User entity|       | Owns: Product    |       | Extends: User    |     |
|    +------------------+       +------------------+       +------------------+     |
+-----------------------------------------------------------------------------------+

# users-subgraph.graphql
type User @key(fields: "id") {
  id: ID!
  email: String!
  displayName: String!
}

# reviews-subgraph.graphql
type Review {
  id: ID!
  rating: Int!
  comment: String!
}

# Extend User entity owned by Users Subgraph
type User @key(fields: "id") {
  id: ID!
  reviews: [Review!]!
}

query {
  user(id: "42") {
    displayName
    reviews { rating comment }
  }
}

const resolvers = {
  User: {
    __resolveReference: async (reference: { id: string }, context: GraphQLContext) => {
      // Rehydrate user reviews using reference.id
      return { id: reference.id };
    },
    reviews: async (parent: { id: string }, _args: unknown, context: GraphQLContext) => {
      return context.db.query('SELECT * FROM reviews WHERE author_id = 1', [parent.id]);
    }
  }
};

Query Plan for:
{
  product(id: "p_101") {
    name                # From Products Subgraph
    shippingCost        # From Shipping Subgraph (Requires: weightGrams from Product)
    reviews { rating }  # From Reviews Subgraph
  }
}

Step 1: Fetch from Products Subgraph
  Query: { product(id: "p_101") { name weightGrams } }

Step 2: Concurrently execute:
  - Step 2a: Query Shipping Subgraph
      Pass Entity: { __typename: "Product", id: "p_101", weightGrams: 540 }
      Fetch: shippingCost
  - Step 2b: Query Reviews Subgraph
      Pass Entity: { __typename: "Product", id: "p_101" }
      Fetch: reviews { rating }

Step 3: Flatten and merge JSON responses; strip 'weightGrams' (unrequested by client); return to user.

query ProductDetails(id: ID!) {
  product(id: id) {
    name                # Subgraph A: Products Service
    inventoryLevel      # Subgraph B: Warehouse Service (Key: sku)
    estimatedDelivery   # Subgraph C: Logistics Service (Requires: warehouseZip, weightGrams)
    reviews(limit: 5) { # Subgraph D: Reviews Service (Key: id)
      rating
    }
  }
}

       [ Router Ingress ]
               |
               v
     [ Subgraph A: 12 ms ]
               |
        +------+------+
        |             |
        v             v
  [ Subgraph D: 15ms ] [ Subgraph B: 18 ms ]
                      |
                      v
                [ Subgraph C: 25 ms ]

// subscription-server.ts
import http from 'node:http';
import express from 'express';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { RedisPubSub } from 'graphql-redis-subscriptions';
import Redis from 'ioredis';
import { schema } from './schema';

const app = express();
const server = http.createServer(app);

// Configure Redis Pub/Sub connections
const redisOptions = {
  host: process.env.REDIS_HOST || 'localhost',
  port: parseInt(process.env.REDIS_PORT || '6379', 10),
  retryStrategy: (times: number) => Math.min(times * 50, 2000)
};

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

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

// Configure graphql-ws server lifecycle
useServer(
  {
    schema,
    // 1. Connection Initialization & Authentication Guard
    onConnect: async (ctx) => {
      const authHeader = ctx.connectionParams?.authToken as string;
      if (!authHeader) {
        // Reject unauthenticated WebSocket connections
        return false;
      }

      try {
        const user = await validateAuthToken(authHeader);
        // Bind user to context
        ctx.extra = { currentUser: user };
        return true;
      } catch {
        return false;
      }
    },

    // 2. Build GraphQL Context for each subscription operation
    context: (ctx) => {
      return {
        pubsub,
        currentUser: (ctx.extra as any).currentUser
      };
    },

    // 3. Error and disconnect logging
    onDisconnect: (_ctx, code, reason) => {
      console.log(`[Subscription WS] Client disconnected ({code}): {reason}`);
    }
  },
  wsServer
);

async function validateAuthToken(token: string) {
  // Token verification logic
  return { id: 'u_102', role: 'USER' };
}

server.listen(4000, () => {
  console.log('[Server] GraphQL Subscription Server running on ws://localhost:4000/graphql');
});

# accounts.graphql
extend schema
  @link(url: "https://specs.apollo.dev/federation/v2.0", import: ["@key", "@shareable"])

type User @key(fields: "id") {
  id: ID!
  username: String!
  email: String!
}

type Query {
  user(id: ID!): User
}

// accounts-resolvers.ts
const usersDb = new Map([
  ['usr_1', { id: 'usr_1', username: 'alice', email: 'alice@devthesis.com' }]
]);

export const accountsResolvers = {
  Query: {
    user: (_p: any, args: { id: string }) => usersDb.get(args.id) || null
  },
  User: {
    // Entity Reference Resolver
    __resolveReference: (reference: { id: string }) => {
      return usersDb.get(reference.id) || null;
    }
  }
};

# reviews.graphql
extend schema
  @link(url: "https://specs.apollo.dev/federation/v2.0", import: ["@key", "@shareable"])

type Review {
  id: ID!
  body: String!
  rating: Int!
  authorId: ID!
}

# Extend User entity owned by Accounts Subgraph
type User @key(fields: "id") {
  id: ID!
  reviews: [Review!]!
}

type Query {
  topReviews(limit: Int = 10): [Review!]!
}

// reviews-resolvers.ts
const reviewsDb = [
  { id: 'rev_1', authorId: 'usr_1', rating: 5, body: 'Outstanding deep architectural guide!' },
  { id: 'rev_2', authorId: 'usr_1', rating: 4, body: 'Very thorough mathematical proofs.' }
];

export const reviewsResolvers = {
  Query: {
    topReviews: (_p: any, args: { limit: number }) => {
      return reviewsDb.slice(0, args.limit);
    }
  },
  User: {
    // Entity Reference Resolver for extended User entity
    __resolveReference: (reference: { id: string }) => {
      // Rehydrate minimal reference
      return { id: reference.id };
    },
    reviews: (parent: { id: string }) => {
      return reviewsDb.filter((rev) => rev.authorId === parent.id);
    }
  }
};