// ❌ Common mistake: treating subscription like a query resolver
const resolvers = {
  Subscription: {
    orderStatusChanged: {
      // Missing subscribe function — this will not work
      resolve: (payload) => payload.orderStatusChanged,
    },
  },
};

// ✅ Correct: both subscribe and resolve are required
const resolvers = {
  Subscription: {
    orderStatusChanged: {
      // 1. subscribe() — called ONCE when the client opens the subscription.
      //    Must return an AsyncIterator. Each value it yields triggers resolve().
      subscribe: async (_, { orderId }, ctx) => {
        // Returns an AsyncIterator that emits { orderStatusChanged: Order }
        // objects whenever an order's status changes
        return ctx.pubsub.subscribe(`ORDER_STATUS:${orderId}`);
      },

      // 2. resolve() — called for EACH value yielded by the AsyncIterator.
      //    Transforms the raw payload into the subscription field's return type.
      resolve: (payload: { orderStatusChanged: Order }) => {
        return payload.orderStatusChanged;
      },
    },
  },
};

// ❌ In-memory PubSub — works in local dev, silently broken in production
import { EventEmitter } from 'events';

class InMemoryPubSub {
  private emitter = new EventEmitter();

  async publish(topic: string, payload: unknown) {
    // Emits only within THIS Node.js process
    // If you have 3 pods, only clients connected to THIS pod receive the event
    this.emitter.emit(topic, payload);
  }

  async subscribe(topic: string): AsyncIterator<unknown> {
    // The listener only fires when publish() is called on this same instance
    return createAsyncIterator(this.emitter, topic);
  }
}

// In a 3-pod deployment:
// Client A connects to Pod 1 (subscribes to ORDER_STATUS:o1 on Pod 1's EventEmitter)
// Client B triggers an order update → request routes to Pod 3
// Pod 3 calls pubsub.publish('ORDER_STATUS:o1') → Pod 3's EventEmitter fires
// Pod 1's EventEmitter NEVER fires → Client A receives NOTHING

// ✅ Redis PubSub — events fan out across all pods
import Redis from 'ioredis';
import { RedisPubSub } from 'graphql-redis-subscriptions';

const publisher = new Redis({ host: 'redis', port: 6379 });
const subscriber = new Redis({ host: 'redis', port: 6379 });

// Note: Redis requires separate publisher and subscriber connections.
// A connection in SUBSCRIBE mode cannot issue normal Redis commands.
const pubsub = new RedisPubSub({
  publisher,
  subscriber,
});

// When Pod 3 publishes:
await pubsub.publish('ORDER_STATUS:o1', { orderStatusChanged: updatedOrder });
// Redis broadcasts to ALL subscribers on ALL pods — Pod 1 receives it ✅

// ✅ Kafka PubSub — durable event log, replay, strict ordering
import { Kafka } from 'kafkajs';

const kafka = new Kafka({ brokers: ['kafka:9092'] });
const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: 'graphql-subscriptions' });

// Kafka advantages over Redis for subscriptions:
// - Event replay: subscribers can resume from a committed offset after restart
// - Strict ordering: partition-level ordering guarantees
// - Durability: events survive broker restarts (with replication)
// - Fan-out: same event can be consumed by multiple consumer groups independently

import { withFilter } from 'graphql-subscriptions';

const resolvers = {
  Subscription: {
    orderStatusChanged: {
      // withFilter wraps the base AsyncIterator
      // The filter function runs for EVERY published event, for EVERY subscriber
      subscribe: withFilter(
        // 1. Base subscription — subscribes to the broad ORDER_STATUS topic
        (_, __, ctx) => ctx.pubsub.subscribe('ORDER_STATUS'),

        // 2. Filter function — called per event per subscriber
        // Return true to deliver to this subscriber, false to skip
        (payload: { orderStatusChanged: Order }, variables: { orderId: string }, ctx) => {
          // Only deliver if the order ID matches what this subscriber requested
          // AND the subscriber has permission to view this order
          return (
            payload.orderStatusChanged.id === variables.orderId &&
            ctx.user.canViewOrder(variables.orderId)
          );
        }
      ),
      resolve: (payload) => payload.orderStatusChanged,
    },
  },
};

// ✅ graphql-ws v5 — the standard WebSocket subscription protocol
import { createServer } from 'http';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { createYoga } from 'graphql-yoga';

const yoga = createYoga({ schema });
const httpServer = createServer(yoga);
const wsServer = new WebSocketServer({ server: httpServer, path: '/graphql' });

useServer(
  {
    schema,
    context: (ctx) => ({
      user: authenticateWsConnection(ctx.connectionParams),
      pubsub,
    }),
    // v5.16+: lazy initialization — connection is not authenticated
    // until the first subscription operation is sent, reducing auth load
    // for clients that open a connection but haven't sent operations yet
    onConnect: async (ctx) => {
      const token = ctx.connectionParams?.authorization;
      if (!token) return false; // Reject unauthenticated connections immediately
    },
  },
  wsServer
);

// ✅ graphql-sse — stateless, HTTP/2, horizontally scalable without sticky sessions
import { createHandler } from 'graphql-sse/lib/use/node';
import { createServer } from 'http';
import { schema } from './schema';

const handler = createHandler({
  schema,
  context: (req) => ({
    user: authenticateRequest(req),
    pubsub,
  }),
});

// SSE uses standard HTTP — no WebSocket upgrade required
// Advantage: works through HTTP proxies, load balancers, and CDNs without configuration
// Advantage: automatic reconnection is handled by the browser's EventSource API
// Tradeoff: unidirectional — client cannot send messages after the connection is open
createServer((req, res) => {
  if (req.method === 'POST' && req.url === '/graphql/stream') {
    return handler(req, res);
  }
  // ... handle other routes
});

# router.yaml — enable subscription support in Apollo Router v1.40+
supergraph:
  listen: 0.0.0.0:4000

subscription:
  enabled: true
  # Router uses SSE to communicate with subgraphs, then re-packages
  # the events in the transport the client requested (WS or SSE)
  mode:
    passthrough:
      all:
        path: /graphql

# Subgraph schema — declare the subscription field in the owning subgraph
type Subscription {
  orderStatusChanged(orderId: ID!): Order!
}