GraphQL و gRPC في الإنتاج: contracts و DataLoader و limits و deadlines

// service.js: catalog service بـ gRPC، و GraphQL gateway فوقه بالـ DataLoader والـ limits والـ errors المتظبّطة
import grpc from "@grpc/grpc-js";
import protoLoader from "@grpc/proto-loader";
import DataLoader from "dataloader";
import { EventEmitter } from "node:events";
import { createServer } from "node:http";
import { buildSchema, parse, validate, execute, specifiedRules, NoSchemaIntrospectionCustomRule, GraphQLError, Kind } from "graphql";

const PROTO = new URL("./catalog.proto", import.meta.url).pathname;
const pkgDef = protoLoader.loadSync(PROTO, { keepCase: false, longs: String, defaults: true });
const { Catalog } = grpc.loadPackageDefinition(pkgDef).catalog.v1;

/* ---------------- الـ gRPC server ---------------- */

const fail = (code, details) => Object.assign(new Error(details), { code, details });
const encodeCursor = (id) => Buffer.from(id).toString("base64url");
const decodeCursor = (c) => {
  const id = Buffer.from(c, "base64url").toString();
  if (!id || encodeCursor(id) !== c) throw fail(grpc.status.INVALID_ARGUMENT, "bad cursor");
  return id;
};

export async function startCatalog({ products = [], port = 0, latencyMs = 0 } = {}) {
  const byId = new Map(products.map((p) => [p.id, { ...p }]));
  const calls = { GetProducts: [], ListProducts: 0, WatchStock: 0 };
  const events = new EventEmitter();
  const later = (call, cb, f) => setTimeout(() => {
    if (call.cancelled) return; // الـ client استسلم (deadline)، ما تكمّلش
    try { cb(null, f()); } catch (e) { cb(e); }
  }, latencyMs);

  const server = new grpc.Server();
  server.addService(Catalog.service, {
    GetProducts(call, cb) {
      const ids = call.request.ids;
      calls.GetProducts.push([...ids]);
      later(call, cb, () => {
        if (ids.length === 0 || ids.length > 100) throw fail(grpc.status.INVALID_ARGUMENT, "ids must be 1..100");
        return { products: ids.map((id) => byId.get(id)).filter(Boolean) };
      });
    },
    ListProducts(call, cb) {
      calls.ListProducts++;
      later(call, cb, () => {
        const { after, sellerId } = call.request;
        const limit = call.request.limit || 10;
        if (limit < 1 || limit > 50) throw fail(grpc.status.INVALID_ARGUMENT, "limit must be 1..50");
        const from = after ? decodeCursor(after) : null;
        const all = [...byId.values()].filter((p) => !sellerId || p.sellerId === sellerId).sort((a, b) => (a.id < b.id ? -1 : 1));
        const rest = from ? all.filter((p) => p.id > from) : all;
        const page = rest.slice(0, limit);
        return { products: page, nextCursor: rest.length > limit ? encodeCursor(page.at(-1).id) : "" };
      });
    },
    WatchStock(call) {
      calls.WatchStock++;
      const ids = new Set(call.request.ids);
      const on = (e) => { if (ids.has(e.id)) call.write(e); };
      events.on("stock", on);
      call.on("cancelled", () => { events.off("stock", on); call.end(); });
    },
  });
  const bound = await new Promise((ok, no) => server.bindAsync(`127.
…

$ node lab.js
— 1) gRPC مباشرة —
  ListProducts: p01 1000 , p02 1250 , p03 1500 ← next cDAz
  WatchStock على p01 بس: [ 'p01=42' ]

— 2) N+1: من غير DataLoader ومعاه —
  واحد واحد: 12 calls في 148ms
  الـ gateway: 1 call لـ GetProducts فيه 12 ids، و getMany للـ sellers 1 مرة بـ 4 ids، في 27ms تمام

— 3) queries بتضر —
  عمق كبير: DEPTH_LIMIT: depth 7 > 5
  كتير أوي: COST_LIMIT: cost 2651 > 150
  introspection: GRAPHQL_VALIDATION_FAILED: GraphQL introspection has been disabled, but the requested query contained the field "__schema". | GRAPHQL_VALIDATION_FAILED: GraphQL introspection has been disabled, but the requested query contained the field "types".
  first برّه الحد: BAD_USER_INPUT: limit must be 1..50

— 4) الـ catalog بطيء —
  رجع بعد 202ms مش ألفين: UPSTREAM_UNAVAILABLE: catalog unavailable ، data: {"product":null}

— 5) الـ sellers DB وقعت —
  الـ client شايف: {"data":{"product":{"name":"منتج 3","price":{"amount":1500,"currency":"EGP"},"seller":null}},"errors":[{"message":"Internal server error","path":["product","seller"],"locations":[{"line":1,"column":55}],"extensions":{"code":"INTERNAL"}}]}
  والـ log عندنا: connect ECONNREFUSED 10.0.3.7:5432 user=app password=s3cret

// gateway.naive.js: GraphQL فوق REST، أول نسخة
const resolvers = {
  Query: {
    products: async (_, { first }) => (await fetch(`${CATALOG}/products?limit=${first}`)).json(), // first مليون؟ ماشي
  },
  Product: {
    // لكل product في الـ list، request للـ seller. خمسين product ← خمسين request
    seller: async (p) => (await fetch(`${SELLERS}/sellers/${p.sellerId}`)).json(), // ومن غير timeout
  },
  Seller: {
    // seller ← products ← seller ← products ... الـ client يقدر يطلب عمق مالوش آخر
    products: async (s) => (await fetch(`${CATALOG}/products?seller=${s.id}`)).json(),
  },
};
// وكمان:
// - الـ errors بتطلع للـ client زي ما هي: connect ECONNREFUSED db.internal:5432
// - الـ introspection مفتوح في الإنتاج، فأي حد يشوف الـ schema كلها
// - JSON بين الـ services: من غير contract، و field اتشال من غير ما حد ياخد باله

syntax = "proto3";

package catalog.v1;

// الـ contract بين الـ gateway والـ catalog service. الأرقام دي ما تتغيرش أبدًا بعد ما تنزل
service Catalog {
  rpc GetProducts (GetProductsRequest) returns (ProductList);
  rpc ListProducts (ListProductsRequest) returns (ProductPage);
  rpc WatchStock (WatchStockRequest) returns (stream StockEvent);
}

message Product {
  string id = 1;
  string name = 2;
  int32 price_cents = 3;
  string seller_id = 4;
  int32 stock = 5;
}

message GetProductsRequest { repeated string ids = 1; }
message ProductList { repeated Product products = 1; }

message ListProductsRequest {
  int32 limit = 1;
  string after = 2;
  string seller_id = 3;
}
message ProductPage {
  repeated Product products = 1;
  string next_cursor = 2;
}

message WatchStockRequest { repeated string ids = 1; }
message StockEvent {
  string id = 1;
  int32 stock = 2;
}

// lab.js: catalog service بـ gRPC، و GraphQL gateway قدامه
import { startCatalog, catalogClient, createGateway } from "./service.js";

const products = Array.from({ length: 20 }, (_, i) => ({ id: `p${String(i + 1).padStart(2, "0")}`, name: `منتج ${i + 1}`, priceCents: 1000 + i * 250, sellerId: `s${(i % 4) + 1}`, stock: i % 5 }));
const catalog = await startCatalog({ products, latencyMs: 10 });
const sellerCalls = [];
const sellers = { getMany: async (ids) => { sellerCalls.push(ids.length); return ids.map((id) => ({ id, name: `محل ${id}` })); } };
const gw = createGateway({ catalogAddr: catalog.address, sellers, maxDepth: 5, maxCost: 150 });
const show = (r) => (r.body.errors ? r.body.errors.map((e) => `${e.extensions.code}: ${e.message}`).join(" | ") : "تمام");

console.log("— 1) gRPC مباشرة —");
const client = catalogClient(catalog.address);
const page = await new Promise((ok, no) => client.ListProducts({ limit: 3 }, { deadline: new Date(Date.now() + 500) }, (e, r) => (e ? no(e) : ok(r))));
console.log("  ListProducts:", page.products.map((p) => `${p.id} ${p.priceCents}`).join(" , "), "← next", page.nextCursor);
const events = [];
const stream = client.WatchStock({ ids: ["p01"] });
stream.on("data", (e) => events.push(e)).on("error", () => {});
await new Promise((r) => setTimeout(r, 100));
catalog.setStock("p01", 42); catalog.setStock("p02", 9);
await new Promise((r) => setTimeout(r, 100));
stream.cancel();
console.log("  WatchStock على p01 بس:", events.map((e) => `${e.id}=${e.stock}`));

console.log("\n— 2) N+1: من غير DataLoader ومعاه —");
const ids = products.slice(0, 12).map((p) => p.id);
const t0 = Date.now();
for (const id of ids) await new Promise((ok) => client.GetProducts({ ids: [id] }, (e, r) => ok(r)));
console.log(`  واحد واحد: ${ids.length} calls في ${Date.now() - t0}ms`);
catalog.calls.GetProducts.length = 0;
const t1 = Date.now();
const q = `{ ${ids.map((id, i) => `p${i}: product(id: "${id}") { name seller { name } }`).join(" ")} }`;
const r2 = await gw.run({ query: q });
console.log(`  الـ gateway: ${catalog.calls.GetProducts.length} call لـ GetProducts فيه ${catalog.calls.GetProducts[0].length} ids، و getMany للـ sellers ${sellerCalls.length} مرة بـ ${sellerCalls[0]} ids، في ${Date.now() - t1}ms`, show(r2));

console.log("\n— 3) queries بتضر —");
const deep = `{ product(id: "p01") { seller { products { seller { products { seller { name } } } } } } }`;
console.log("  عمق كبير:", show(await gw.run({ query: deep })));
console.log("  كتير أوي:", show(await gw.run({ query: `{ products(first: 50) { items { seller { products(first: 50) { name } } } } }` })));
console.log("  introspection:", show(await gw.run({ query: `{ __schema { types { name } } }` })));
console.log("  first برّه الحد:", show(await gw.run({ query: `{ products(first: 80) { nextCursor } }` })));

console.log("\n— 4) الـ catalog بطيء —");
const slow = await startCatalog({ products, latencyMs: 2000 });
const gwSlow = createGateway({ catalogAddr: slow.address, sellers, t
…