index.ts

/**
 * index.ts — Orders subgraph entry point.
 *
 * Apollo Server v5 + Express 5 + Manual HTTP Callback Protocol (Redis-backed).
 *
 * ─────────────────────────────────────────────────────────────────────────────
 * Architecture contrast with the Notifications subgraph:
 *
 *   Notifications: uses ApolloServerPluginSubscriptionCallback
 *     - Subscription state lives IN PROCESS (SubscriptionManager in-memory)
 *     - Only the pod that received the init can deliver events
 *     - Simple, but NOT horizontally scalable for subscriptions
 *
 *   Orders (this file): manual callback protocol with Redis state
 *     - Subscription state lives in REDIS (shared across all pods)
 *     - ANY pod can service ANY Kafka event — queries Redis to find matching subs
 *     - Heartbeat is handled by the subscription-manager service (not this pod)
 *     - True stateless horizontal scaling: add pods → add Kafka consumer capacity
 * ─────────────────────────────────────────────────────────────────────────────
 *
 * Request handling overview:
 *
 *   POST /graphql  (Accept: application/json;callbackSpec=1.0)
 *     ↓  subscriptionCallbackMiddleware (BEFORE Apollo Server)
 *     ↓  Parses extensions.subscription.{callbackUrl, subscriptionId, verifier}
 *     ↓  Stores in Redis → returns {"data": null}
 *     ↓  (Apollo Server never sees this request)
 *
 *   POST /graphql  (regular query/mutation)
 *     ↓  Apollo Server expressMiddleware
 *     ↓  Resolves via resolvers.ts
 *     ↓  updateOrderStatus publishes to Kafka
 *
 *   Kafka consumer loop (background):
 *     ↓  order-status-changed event
 *     ↓  getSubscriptionsByOrderId(orderId) → Redis SMEMBERS
 *     ↓  For each subscriptionId: getSubscription → callbackUrl, verifier
 *     ↓  POST {type: "next", id, payload: {data: {orderStatusChanged: {...}}}} to callbackUrl
 *     ↓  204 OK → log success
 *     ↓  404    → deleteSubscription (client gone)
 *     ↓  5xx    → log warn (at-most-once; see note below)
 *
 * Delivery semantics note:
 *   This demo uses at-most-once delivery per subscription. If the Router returns 5xx,
 *   we log but do not NACK the Kafka message (which would cause redelivery to ALL
 *   subscribers, including ones that already received it). Production systems should
 *   use per-subscription offset tracking or idempotency keys to safely retry.
 */

import { readFileSync } from "fs";
import { fileURLToPath } from "url";
import { dirname, join } from "path";
import { ApolloServer } from "@apollo/server";
import { expressMiddleware } from "@as-integrations/express5";
import { buildSubgraphSchema } from "@apollo/subgraph";
import gql from "graphql-tag";
import express, { type Request, type Response, type NextFunction } from "express";
import cors from "cors";
import { resolvers, type OrderStatusEvent } from "./resolvers.js";
import {
  connectRedis,
  disconnectRedis,
  redis,
  storeSubscription,
  getSubscription,
  getSubscriptionsByOrderId,
  deleteSubscription,
} from "./redis.js";
import {
  getProducer,
  createConsumer,
  disconnectKafka,
  TOPIC_ORDER_STATUS_CHANGED,
} from "./kafka.js";
import type { Consumer } from "kafkajs";

// ── Load schema ──
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);
const schemaSource = readFileSync(join(__dirname, "schema.graphql"), "utf-8");
const typeDefs = (gql as unknown as typeof gql.default)(schemaSource);

// ── Build federated subgraph schema ──
const schema = buildSubgraphSchema({ typeDefs, resolvers });

// ── Apollo Server (handles queries + mutations only) ──
// No subscription plugin — subscription initiations are handled by the
// subscriptionCallbackMiddleware BEFORE this middleware runs.
const server = new ApolloServer({
  schema,
  introspection: true,
});

// ── Subscription init middleware ──
// Intercepts subscription init requests (Router sends Accept: application/json;callbackSpec=1.0).
// Stores subscription state in Redis and returns {"data": null} to acknowledge.
async function subscriptionCallbackMiddleware(
  req: Request,
  res: Response,
  next: NextFunction
): Promise<void> {
  const acceptHeader = req.headers["accept"] ?? "";
  if (!acceptHeader.includes("callbackSpec=1.0")) {
    // Not a subscription init — let Apollo Server handle it
    return next();
  }

  const body = req.body as {
    query?: string;
    variables?: Record<string, unknown>;
    extensions?: {
      subscription?: {
        callbackUrl?: string;
        subscriptionId?: string;
        verifier?: string;
      };
    };
  };

  const sub = body.extensions?.subscription;
  if (!sub?.callbackUrl || !sub?.subscriptionId || !sub?.verifier) {
    res.status(400).json({
      errors: [{ message: "Missing required subscription extension fields" }],
    });
    return;
  }

  // Extract orderId from subscription variables
  const orderId = String(body.variables?.["orderId"] ?? "");
  if (!orderId) {
    res.status(400).json({
      errors: [{ message: "orderStatusChanged requires variable: orderId" }],
    });
    return;
  }

  await storeSubscription(sub.subscriptionId, {
    callbackUrl: sub.callbackUrl,
    verifier: sub.verifier,
    orderId,
    indexKey: `subindex:${orderId}`,
    subgraph: "orders",
    variables: JSON.stringify(body.variables ?? {}),
    createdAt: new Date().toISOString(),
  });

  console.log(
    `🔔 Subscription init: id=${sub.subscriptionId} orderId=${orderId} callbackUrl=${sub.callbackUrl}`
  );

  // Acknowledge to the Router: subscription registered, events will follow
  res.status(200).json({ data: null });
}

// ── Kafka consumer: deliver events to Router callback URLs ──
let consumer: Consumer | null = null;

async function startEventDeliveryLoop(): Promise<void> {
  consumer = createConsumer("orders-subgraph-group");
  await consumer.connect();
  console.log("✅ Kafka event consumer connected");

  await consumer.subscribe({
    topics: [TOPIC_ORDER_STATUS_CHANGED],
    fromBeginning: false,
  });

  await consumer.run({
    eachMessage: async ({ topic: _topic, partition, message }) => {
      if (!message.value) return;

      let event: OrderStatusEvent;
      try {
        event = JSON.parse(message.value.toString()) as OrderStatusEvent;
      } catch (err) {
        console.error("❌ Failed to parse Kafka message:", err);
        return; // Discard unparseable messages — don't block partition
      }

      console.log(
        `📨 Kafka event: orderId=${event.orderId} ${event.previousStatus} → ${event.newStatus} (partition=${partition})`
      );

      // Find all active subscriptions interested in this orderId
      const subIds = await getSubscriptionsByOrderId(event.orderId);
      if (subIds.length === 0) return;

      console.log(
        `📬 Delivering event to ${subIds.length} subscriber(s) for orderId=${event.orderId}`
      );

      // Deliver to each subscriber — fire-and-forget per-subscriber errors
      // (at-most-once; see module docstring for production considerations)
      for (const subId of subIds) {
        const sub = await getSubscription(subId);
        if (!sub) {
          // Subscription expired from Redis — remove stale index entry
          await redis.sRem(`subindex:${event.orderId}`, subId);
          continue;
        }

        const payload = {
          kind: "subscription",
          action: "next",
          id: subId,
          verifier: sub.verifier,
          payload: {
            data: {
              orderStatusChanged: {
                orderId: event.orderId,
                previousStatus: event.previousStatus,
                newStatus: event.newStatus,
                timestamp: event.timestamp,
                order: event.order ?? null,
              },
            },
          },
        };

        try {
          const res = await fetch(sub.callbackUrl, {
            method: "POST",
            headers: {
              "Content-Type": "application/json",
              "subscription-protocol": "callback/1.0",
            },
            body: JSON.stringify(payload),
            signal: AbortSignal.timeout(10_000),
          });

          if (res.status === 200 || res.status === 204) {
            console.log(`✅ Event delivered: ${subId}`);

            // If the order reached a terminal state, send "complete" to close the subscription
            if (
              event.newStatus === "DELIVERED" ||
              event.newStatus === "CANCELLED"
            ) {
              await sendComplete(subId, sub.callbackUrl, sub.verifier, sub.indexKey);
            }
          } else if (res.status === 404) {
            // Client disconnected — Router has cleaned up its side
            console.log(`🗑️  Delivery 404 — client gone, cleaning up: ${subId}`);
            await deleteSubscription(subId, sub.indexKey);
          } else {
            console.warn(
              `⚠️  Delivery ${res.status} for sub ${subId} — skipping (at-most-once)`
            );
          }
        } catch (err) {
          console.warn(
            `⚠️  Delivery network error for sub ${subId}:`,
            (err as Error).message
          );
        }
      }
    },
  });

  console.log(
    `✅ Event delivery loop started — listening on [${TOPIC_ORDER_STATUS_CHANGED}]`
  );
}

/**
 * Send a "complete" message to the Router callback URL to gracefully close
 * a subscription when the order reaches a terminal state (DELIVERED / CANCELLED).
 */
async function sendComplete(
  subscriptionId: string,
  callbackUrl: string,
  verifier: string,
  indexKey: string
): Promise<void> {
  try {
    await fetch(callbackUrl, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "subscription-protocol": "callback/1.0",
      },
      body: JSON.stringify({
        kind: "subscription",
        action: "complete",
        id: subscriptionId,
        verifier,
      }),
      signal: AbortSignal.timeout(5_000),
    });
    console.log(
      `✅ Sent complete for sub ${subscriptionId} (order reached terminal state)`
    );
  } catch (err) {
    console.warn(
      `⚠️  Failed to send complete for ${subscriptionId}:`,
      (err as Error).message
    );
  } finally {
    // Wrap separately so a Redis failure here doesn't propagate out of sendComplete
    // and crash the Kafka consumer loop (which would stop all event delivery).
    try {
      await deleteSubscription(subscriptionId, indexKey);
    } catch (cleanupErr) {
      console.warn(
        `⚠️  Failed to clean up subscription ${subscriptionId} after complete:`,
        (cleanupErr as Error).message
      );
    }
  }
}

// ── Bootstrap ──
async function main() {
  const PORT = parseInt(process.env.PORT ?? "4003", 10);

  // 1. Connect Redis (subscription state store)
  await connectRedis();

  // 2. Start Kafka producer (used by mutations) — warm it up early
  await getProducer();

  // 3. Start Kafka event delivery loop (Kafka → Redis lookup → Router callback)
  await startEventDeliveryLoop().catch((err) => {
    console.warn(
      "⚠️  Kafka event consumer failed to start:",
      (err as Error).message
    );
    console.warn("   Subscriptions will NOT receive events until Kafka is available.");
  });

  // 4. Start Apollo Server (handles queries + mutations)
  await server.start();

  // 5. Create Express app
  const app = express();

  // Health check (for Docker/K8s probes)
  app.get("/health", (_req, res) => {
    res.json({ status: "ok" });
  });

  // Mount middleware for /graphql:
  //   a. JSON body parser (must run before our interceptor reads req.body)
  //   b. subscriptionCallbackMiddleware (intercepts subscription inits)
  //   c. Apollo Server (handles queries + mutations)
  app.use(
    "/graphql",
    cors<cors.CorsRequest>(),
    express.json(),
    subscriptionCallbackMiddleware,
    expressMiddleware(server)
  );

  // 6. Start listening
  const httpServer = app.listen(PORT, () => {
    console.log(`🚀 Orders subgraph ready at http://localhost:${PORT}/graphql`);
    console.log(`💚 Health check at http://localhost:${PORT}/health`);
    console.log(`🏗️  Architecture: Redis-backed subscription state (stateless pods)`);
  });

  // ── Graceful shutdown ──
  const shutdown = async (signal: string) => {
    console.log(`\n${signal} received — shutting down gracefully...`);
    httpServer.close();
    await server.stop();
    if (consumer) await consumer.disconnect();
    await disconnectKafka();
    await disconnectRedis();
    console.log("👋 Orders subgraph stopped.");
    process.exit(0);
  };

  process.on("SIGTERM", () => shutdown("SIGTERM"));
  process.on("SIGINT", () => shutdown("SIGINT"));
}

main().catch((err) => {
  console.error("❌ Failed to start orders subgraph:", err);
  process.exit(1);
});