Claude Code + gRPC微服务通信：从Protobuf到流式RPC全链路实战

请用表格对比 gRPC 的四种通信模式：Unary RPC、Server Streaming、
Client Streaming、Bidirectional Streaming。包含：定义、适用场景、
数据流向、类比。

┌──────────────────┬────────────────────┬───────────────────┐
│ 模式              │ 数据流向            │ 典型场景           │
├──────────────────┼────────────────────┼───────────────────┤
│ Unary RPC        │ 请求→响应（一对一）  │ 用户登录、查订单   │
│ Server Streaming │ 请求→流式响应       │ 实时日志、股票行情  │
│ Client Streaming │ 流式请求→响应       │ 文件上传、批量写入  │
│ Bidi Streaming   │ 双向流式通信        │ 聊天、实时协作     │
└──────────────────┴────────────────────┴───────────────────┘

mkdir grpc-order-service && cd grpc-order-service
claude

请为一个 gRPC 订单微服务创建 Node.js 项目结构，要求：
1. 使用 TypeScript
2. 包含 proto 定义目录 proto/
3. 生成的代码目录 src/generated/
4. 服务端代码 src/server/
5. 客户端代码 src/client/
6. 共享工具 src/shared/
7. 配置好 tsconfig.json 和 package.json
8. 添加 @grpc/grpc-js、@grpc/proto-loader、protobufjs 依赖

grpc-order-service/
├── proto/
│   └── order.proto          # Protobuf 服务定义
├── src/
│   ├── generated/            # 自动生成的代码（不手动编辑）
│   ├── server/
│   │   ├── index.ts          # 服务入口
│   │   ├── orderService.ts   # 订单服务实现
│   │   └── interceptors.ts   # 拦截器
│   ├── client/
│   │   └── index.ts          # 客户端调用示例
│   └── shared/
│       └── errors.ts         # 统一错误码
├── tests/
│   └── order.test.ts         # 单元测试
├── package.json
├── tsconfig.json
└── CLAUDE.md                 # Claude Code 项目上下文

npm install @grpc/grpc-js @grpc/proto-loader protobufjs
npm install -D typescript @types/node tsx vitest

请为电商订单服务编写 proto/order.proto 文件，要求：
1. 包名 order.v1
2. 服务 OrderService 包含以下 RPC：
   - CreateOrder（Unary）：创建订单
   - GetOrder（Unary）：查询订单
   - ListOrders（Server Streaming）：流式返回订单列表
   - UploadOrderItems（Client Streaming）：批量上传订单商品
   - OrderChat（Bidirectional Streaming）：订单实时讨论
3. 定义完整的 Message：Order、OrderItem、CreateOrderRequest/Response 等
4. 包含字段验证注释和字段编号规划
5. 使用 google.protobuf.Timestamp 和 Empty

syntax = "proto3";

package order.v1;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

option go_package = "github.com/example/order/v1;orderv1";

// 订单状态枚举
enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PENDING = 1;
  ORDER_STATUS_CONFIRMED = 2;
  ORDER_STATUS_SHIPPED = 3;
  ORDER_STATUS_DELIVERED = 4;
  ORDER_STATUS_CANCELLED = 5;
}

// 订单商品
message OrderItem {
  string product_id = 1;
  string product_name = 2;
  int32 quantity = 3;
  double unit_price = 4;
}

// 订单主体
message Order {
  string order_id = 1;
  string user_id = 2;
  repeated OrderItem items = 3;
  OrderStatus status = 4;
  double total_amount = 5;
  string shipping_address = 6;
  google.protobuf.Timestamp created_at = 7;
  google.protobuf.Timestamp updated_at = 8;
}

// === Unary RPC: CreateOrder ===
message CreateOrderRequest {
  string user_id = 1;
  repeated OrderItem items = 2;
  string shipping_address = 3;
}

message CreateOrderResponse {
  Order order = 1;
}

// === Unary RPC: GetOrder ===
message GetOrderRequest {
  string order_id = 1;
}

message GetOrderResponse {
  Order order = 1;
}

// === Server Streaming RPC: ListOrders ===
message ListOrdersRequest {
  string user_id = 1;
  OrderStatus filter_status = 2;  // 可选状态过滤
}

// 流式返回，每条消息即一个订单
message ListOrdersResponse {
  Order order = 1;
}

// === Client Streaming RPC: UploadOrderItems ===
message UploadOrderItemRequest {
  string order_id = 1;
  OrderItem item = 2;
}

message UploadOrderItemsResponse {
  string order_id = 1;
  int32 total_items = 2;
  double total_amount = 3;
}

// === Bidi Streaming RPC: OrderChat ===
message OrderChatMessage {
  string order_id = 1;
  string sender = 2;
  string message = 3;
  google.protobuf.Timestamp sent_at = 4;
}

// 订单服务定义
service OrderService {
  // Unary: 创建订单
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse);

  // Unary: 查询订单
  rpc GetOrder(GetOrderRequest) returns (GetOrderResponse);

  // Server Streaming: 流式列出用户订单
  rpc ListOrders(ListOrdersRequest) returns (stream ListOrdersResponse);

  // Client Streaming: 批量上传订单商品
  rpc UploadOrderItems(stream UploadOrderItemRequest) returns (UploadOrderItemsResponse);

  // Bidi Streaming: 订单实时讨论
  rpc OrderChat(stream OrderChatMessage) returns (stream OrderChatMessage);
}

请编写一个脚本 scripts/generate.sh，功能：
1. 使用 grpc_tools_node_protoc 生成 TypeScript 类型定义
2. 使用 grpc_tools_node_protoc_ts 生成服务桩
3. 输出到 src/generated/ 目录
4. 同时生成客户端和服务端存根

#!/bin/bash
set -e

PROTO_DIR="proto"
OUT_DIR="src/generated"

mkdir -p "$OUT_DIR"

# 安装生成工具
npm install -D grpc-tools grpc-tools-ts

# 生成 JS 代码和 TypeScript 类型
npx grpc_tools_node_protoc \
  --js_out=import_style=commonjs,binary:"$OUT_DIR" \
  --grpc_out=grpc_js:"$OUT_DIR" \
  --plugin=protoc-gen-grpc=$(npm bin)/grpc_tools_node_protoc_plugin \
  -I "$PROTO_DIR" \
  "$PROTO_DIR"/*.proto

# 生成 TypeScript d.ts 类型声明
npx grpc_tools_node_protoc \
  --plugin=protoc-gen-ts=$(npm bin)/protoc-gen-ts \
  --ts_out=grpc_js:"$OUT_DIR" \
  -I "$PROTO_DIR" \
  "$PROTO_DIR"/*.proto

echo "✅ Proto 代码生成完成: $OUT_DIR/"

chmod +x scripts/generate.sh
./scripts/generate.sh

src/generated/
├── order_pb.js          # Protobuf 消息序列化/反序列化
├── order_pb.d.ts        # TypeScript 类型定义
├── OrderService_grpc_pb.js   # gRPC 服务端/客户端存根
└── OrderService_grpc_pb.d.ts

请基于生成的 gRPC 存根，实现 src/server/orderService.ts：
1. 实现 CreateOrder 和 GetOrder 两个 Unary RPC
2. 使用内存 Map 模拟数据存储
3. 生成 UUID 作为订单 ID
4. 包含完整的错误处理（NOT_FOUND、INVALID_ARGUMENT）
5. 添加日志记录

import * as grpc from '@grpc/grpc-js';
import { v4 as uuidv4 } from 'uuid';
import {
  OrderServiceService,
  IOrderServiceServer,
} from '../generated/OrderService_grpc_pb';
import {
  Order,
  OrderItem,
  CreateOrderRequest,
  CreateOrderResponse,
  GetOrderRequest,
  GetOrderResponse,
  ListOrdersRequest,
  ListOrdersResponse,
  UploadOrderItemRequest,
  UploadOrderItemsResponse,
  OrderChatMessage,
} from '../generated/order_pb';

// 内存数据存储
const orders = new Map<string, Order>();

// === Unary RPC: CreateOrder ===
function createOrder(
  call: grpc.ServerUnaryCall<CreateOrderRequest, CreateOrderResponse>,
  callback: grpc.sendUnaryData<CreateOrderResponse>
) {
  const req = call.request;
  const userId = req.getUserId();
  const items = req.getItemsList();
  const address = req.getShippingAddress();

  // 参数校验
  if (!userId) {
    return callback({
      code: grpc.status.INVALID_ARGUMENT,
      message: 'user_id is required',
    } as grpc.ServiceError);
  }

  if (items.length === 0) {
    return callback({
      code: grpc.status.INVALID_ARGUMENT,
      message: 'order must contain at least one item',
    } as grpc.ServiceError);
  }

  // 计算总金额
  const totalAmount = items.reduce(
    (sum, item) => sum + item.getQuantity() * item.getUnitPrice(),
    0
  );

  // 构建订单
  const order = new Order();
  order.setOrderId(uuidv4());
  order.setUserId(userId);
  order.setItemsList(items);
  order.setStatus(1); // PENDING
  order.setTotalAmount(totalAmount);
  order.setShippingAddress(address);
  order.setCreatedAt(new Date().toISOString());
  order.setUpdatedAt(new Date().toISOString());

  // 存储
  orders.set(order.getOrderId(), order);

  console.log(`[CreateOrder] 订单创建成功: ${order.getOrderId()}, 金额: ¥${totalAmount}`);

  const response = new CreateOrderResponse();
  response.setOrder(order);
  callback(null, response);
}

// === Unary RPC: GetOrder ===
function getOrder(
  call: grpc.ServerUnaryCall<GetOrderRequest, GetOrderResponse>,
  callback: grpc.sendUnaryData<GetOrderResponse>
) {
  const orderId = call.request.getOrderId();

  if (!orderId) {
    return callback({
      code: grpc.status.INVALID_ARGUMENT,
      message: 'order_id is required',
    } as grpc.ServiceError);
  }

  const order = orders.get(orderId);
  if (!order) {
    return callback({
      code: grpc.status.NOT_FOUND,
      message: `Order not found: ${orderId}`,
    } as grpc.ServiceError);
  }

  console.log(`[GetOrder] 查询订单: ${orderId}`);

  const response = new GetOrderResponse();
  response.setOrder(order);
  callback(null, response);
}

// 导出服务定义
export const orderServiceHandlers: IOrderServiceServer = {
  createOrder,
  getOrder,
  // 流式 RPC 在下一步实现
} as unknown as IOrderServiceServer;

import * as grpc from '@grpc/grpc-js';
import { OrderServiceService } from '../generated/OrderService_grpc_pb';
import { orderServiceHandlers } from './orderService';

const PORT = process.env.GRPC_PORT || '50051';

const server = new grpc.Server();

// 注册服务
server.addService(OrderServiceService, orderServiceHandlers as any);

// 启动
server.bindAsync(
  `0.0.0.0:${PORT}`,
  grpc.ServerCredentials.createInsecure(),
  (err, port) => {
    if (err) {
      console.error('服务启动失败:', err);
      process.exit(1);
    }
    console.log(`✅ gRPC OrderService 运行在端口 ${port}`);
    server.start();
  }
);

// 优雅关闭
process.on('SIGTERM', () => {
  console.log('正在关闭 gRPC 服务...');
  server.tryShutdown(() => {
    console.log('服务已关闭');
    process.exit(0);
  });
});

npx tsx src/server/index.ts
# 输出: ✅ gRPC OrderService 运行在端口 50051

# 列出服务（需启用 reflection，后文实现）
grpcurl -plaintext localhost:50051 list

# 创建订单
grpcurl -plaintext -d '{
  "user_id": "u-001",
  "items": [{"product_id": "p-001", "product_name": "键盘", "quantity": 2, "unit_price": 299}],
  "shipping_address": "北京市海淀区"
}' localhost:50051 order.v1.OrderService/CreateOrder

请在 src/server/orderService.ts 中补充实现以下流式 RPC：
1. ListOrders（Server Streaming）: 按用户ID查询订单，每条订单单独推送
2. UploadOrderItems（Client Streaming）: 接收客户端流式上传的商品，汇总返回
3. OrderChat（Bidirectional Streaming）: 双向实时消息通信

要求：
- Server Streaming 每推送一条调用 call.write()，注意 call.emit('error') 处理
- Client Streaming 在 call.on('data') 中累积，call.on('end') 中汇总返回
- Bidi Streaming 用 call.on('data') + call.write() 实现实时交互
- 所有流式 RPC 都要有超时和错误处理

import { Timestamp } from 'google-protobuf/google/protobuf/timestamp_pb';

// === Server Streaming RPC: ListOrders ===
function listOrders(call: grpc.ServerWritableStream<ListOrdersRequest, ListOrdersResponse>) {
  const userId = call.request.getUserId();
  const filterStatus = call.request.getFilterStatus();

  console.log(`[ListOrders] 流式查询订单: userId=${userId}, status=${filterStatus}`);

  let count = 0;
  orders.forEach((order) => {
    // 过滤条件
    if (userId && order.getUserId() !== userId) return;
    if (filterStatus && order.getStatus() !== filterStatus) return;

    const response = new ListOrdersResponse();
    response.setOrder(order);

    try {
      call.write(response);
      count++;
    } catch (err) {
      console.error('[ListOrders] 写入流失败:', err);
      call.emit('error', {
        code: grpc.status.INTERNAL,
        message: 'Stream write failed',
      });
      return;
    }
  });

  console.log(`[ListOrders] 共推送 ${count} 条订单`);
  call.end();
}

// === Client Streaming RPC: UploadOrderItems ===
function uploadOrderItems(
  call: grpc.ServerReadableStream<UploadOrderItemRequest, UploadOrderItemsResponse>,
  callback: grpc.sendUnaryData<UploadOrderItemsResponse>
) {
  const uploadedItems: OrderItem[] = [];
  let targetOrderId = '';

  call.on('data', (request: UploadOrderItemRequest) => {
    targetOrderId = request.getOrderId() || targetOrderId;
    const item = request.getItem();
    if (item) {
      uploadedItems.push(item);
      console.log(`[UploadOrderItems] 接收商品: ${item.getProductName()}`);
    }
  });

  call.on('error', (err) => {
    console.error('[UploadOrderItems] 流错误:', err);
    callback({
      code: grpc.status.INTERNAL,
      message: err.message,
    } as grpc.ServiceError);
  });

  call.on('end', () => {
    // 更新订单
    const order = orders.get(targetOrderId);
    if (!order) {
      return callback({
        code: grpc.status.NOT_FOUND,
        message: `Order not found: ${targetOrderId}`,
      } as grpc.ServiceError);
    }

    const existingItems = order.getItemsList();
    order.setItemsList([...existingItems, ...uploadedItems]);

    const totalAmount = order.getItemsList().reduce(
      (sum, item) => sum + item.getQuantity() * item.getUnitPrice(), 0
    );
    order.setTotalAmount(totalAmount);
    order.setUpdatedAt(new Date().toISOString());
    orders.set(targetOrderId, order);

    const response = new UploadOrderItemsResponse();
    response.setOrderId(targetOrderId);
    response.setTotalItems(order.getItemsList().length);
    response.setTotalAmount(totalAmount);

    console.log(`[UploadOrderItems] 上传完成: ${uploadedItems.length} 件商品`);
    callback(null, response);
  });
}

// === Bidirectional Streaming RPC: OrderChat ===
function orderChat(call: grpc.ServerDuplexStream<OrderChatMessage, OrderChatMessage>) {
  console.log('[OrderChat] 双向流已建立');

  call.on('data', (msg: OrderChatMessage) => {
    const orderId = msg.getOrderId();
    const sender = msg.getSender();
    const text = msg.getMessage();

    console.log(`[OrderChat] ${sender}@${orderId}: ${text}`);

    // 回传确认消息
    const reply = new OrderChatMessage();
    reply.setOrderId(orderId);
    reply.setSender('system');
    reply.setMessage(`[已收到] ${sender}: ${text}`);

    const ts = new Timestamp();
    ts.fromDate(new Date());
    reply.setSentAt(ts);

    call.write(reply);
  });

  call.on('error', (err) => {
    console.error('[OrderChat] 流错误:', err);
  });

  call.on('end', () => {
    console.log('[OrderChat] 客户端关闭连接');
    call.end();
  });
}

请编写 src/server/interceptors.ts，实现以下 gRPC 拦截器：
1. LoggingInterceptor: 记录每个 RPC 调用的方法名、耗时、状态码
2. AuthInterceptor: 从 metadata 中提取 token 并校验
3. ErrorInterceptor: 统一捕获异常并转换为标准 gRPC 状态码
使用 @grpc/grpc-js 的 ServerInterceptor 接口

import * as grpc from '@grpc/grpc-js';
import { Status } from '@grpc/grpc-js/build/src/constants';

// 日志拦截器
export const loggingInterceptor: grpc.ServerInterceptor = (methodDescriptor, callStream) => {
  const methodName = `/${methodDescriptor.service.serviceName}/${methodDescriptor.methodName}`;
  const startTime = Date.now();

  const listener = new grpc.ServerListenerBuilder()
    .withOnReceiveMetadata((metadata, next) => {
      console.log(`[gRPC] --> ${methodName}`);
      next(metadata);
    })
    .withOnReceiveMessage((message, next) => {
      next(message);
    })
    .withOnHalfClose((next) => {
      next();
    })
    .withOnCancel((next) => {
      const duration = Date.now() - startTime;
      console.log(`[gRPC] <-- ${methodName} CANCELLED (${duration}ms)`);
      next();
    })
    .build();

  const responder = new grpc.ServerResponseBuilder()
    .withSendMetadata((metadata, next) => {
      next(metadata);
    })
    .withSendMessage((message, next) => {
      const duration = Date.now() - startTime;
      console.log(`[gRPC] <-- ${methodName} OK (${duration}ms)`);
      next(message);
    })
    .withSendStatus((status, next) => {
      const duration = Date.now() - startTime;
      console.log(
        `[gRPC] <-- ${methodName} ${status.code === Status.OK ? 'OK' : `ERROR(${status.code})`} (${duration}ms)`
      );
      next(status);
    })
    .build();

  return { listener, responder };
};

// 认证拦截器
export const authInterceptor: grpc.ServerInterceptor = (methodDescriptor, callStream) => {
  const methodName = methodDescriptor.methodName;

  // 健康检查无需认证
  const publicMethods = ['Health/Check', 'Health/Watch'];
  const isPublic = publicMethods.some((m) => methodName.includes(m));

  const listener = new grpc.ServerListenerBuilder()
    .withOnReceiveMetadata((metadata, next) => {
      if (isPublic) return next(metadata);

      const token = metadata.get('authorization')[0] as string;
      if (!token || !token.startsWith('Bearer ')) {
        callStream.sendStatus({
          code: grpc.status.UNAUTHENTICATED,
          details: 'Missing or invalid authorization token',
        });
        return;
      }

      // 简化认证逻辑（生产环境应校验 JWT）
      const authToken = token.replace('Bearer ', '');
      if (authToken !== 'valid-secret-token') {
        callStream.sendStatus({
          code: grpc.status.UNAUTHENTICATED,
          details: 'Invalid token',
        });
        return;
      }

      next(metadata);
    })
    .build();

  return { listener, responder: new grpc.ServerResponseBuilder().build() };
};

// 在 src/server/index.ts 中修改
const server = new grpc.Server({
  interceptors: [loggingInterceptor, authInterceptor],
});

请编写 src/client/index.ts，实现对 OrderService 的完整客户端调用：
1. Unary 调用：CreateOrder 和 GetOrder
2. Server Streaming 调用：ListOrders
3. Client Streaming 调用：UploadOrderItems
4. Bidi Streaming 调用：OrderChat
每种模式都要有清晰的日志输出

import * as grpc from '@grpc/grpc-js';
import { OrderServiceClient } from '../generated/OrderService_grpc_pb';
import {
  CreateOrderRequest,
  OrderItem,
  GetOrderRequest,
  ListOrdersRequest,
  UploadOrderItemRequest,
  OrderChatMessage,
} from '../generated/order_pb';

const client = new OrderServiceClient(
  'localhost:50051',
  grpc.credentials.createInsecure()
);

// 添加认证 metadata
const metadata = new grpc.Metadata();
metadata.set('authorization', 'Bearer valid-secret-token');

// === 1. Unary: 创建订单 ===
function testCreateOrder(): Promise<string> {
  return new Promise((resolve, reject) => {
    const item = new OrderItem();
    item.setProductId('p-001');
    item.setProductName('机械键盘');
    item.setQuantity(2);
    item.setUnitPrice(399);

    const request = new CreateOrderRequest();
    request.setUserId('u-001');
    request.setItemsList([item]);
    request.setShippingAddress('北京市海淀区中关村大街1号');

    client.createOrder(request, metadata, (err, response) => {
      if (err) {
        console.error('创建订单失败:', err.message);
        return reject(err);
      }
      const order = response!.getOrder()!;
      console.log(`✅ 订单创建成功: ${order.getOrderId()}, 金额: ¥${order.getTotalAmount()}`);
      resolve(order.getOrderId());
    });
  });
}

// === 2. Server Streaming: 列出订单 ===
function testListOrders(userId: string) {
  const request = new ListOrdersRequest();
  request.setUserId(userId);

  const stream = client.listOrders(request, metadata);

  console.log('📦 开始接收订单流...');

  stream.on('data', (response) => {
    const order = response.getOrder()!;
    console.log(`  - 订单 ${order.getOrderId()}: ¥${order.getTotalAmount()}`);
  });

  stream.on('end', () => {
    console.log('📦 订单流结束');
  });

  stream.on('error', (err) => {
    console.error('流错误:', err.message);
  });
}

// === 3. Client Streaming: 批量上传商品 ===
function testUploadItems(orderId: string): Promise<void> {
  return new Promise((resolve) => {
    const stream = client.uploadOrderItems(metadata, (err, response) => {
      if (err) {
        console.error('上传失败:', err.message);
        return resolve();
      }
      console.log(`✅ 上传完成: ${response!.getTotalItems()} 件, 总计 ¥${response!.getTotalAmount()}`);
      resolve();
    });

    // 流式发送 3 件商品
    const products = [
      { id: 'p-010', name: '显示器', qty: 1, price: 2999 },
      { id: 'p-011', name: '鼠标', qty: 3, price: 149 },
      { id: 'p-012', name: '鼠标垫', qty: 2, price: 49 },
    ];

    products.forEach((p, i) => {
      const item = new OrderItem();
      item.setProductId(p.id);
      item.setProductName(p.name);
      item.setQuantity(p.qty);
      item.setUnitPrice(p.price);

      const request = new UploadOrderItemRequest();
      request.setOrderId(orderId);
      request.setItem(item);

      stream.write(request);
      console.log(`  ↑ 发送商品: ${p.name}`);
    });

    stream.end();
  });
}

// === 4. Bidi Streaming: 实时聊天 ===
function testOrderChat(orderId: string) {
  const stream = client.orderChat(metadata);

  stream.on('data', (msg: OrderChatMessage) => {
    console.log(`  ← ${msg.getSender()}: ${msg.getMessage()}`);
  });

  stream.on('end', () => {
    console.log('聊天结束');
  });

  // 发送 3 条消息
  const messages = ['请问什么时候发货？', '能改地址吗？', '谢谢！'];
  messages.forEach((text, i) => {
    setTimeout(() => {
      const msg = new OrderChatMessage();
      msg.setOrderId(orderId);
      msg.setSender('customer');
      msg.setMessage(text);
      stream.write(msg);
      console.log(`  → customer: ${text}`);

      if (i === messages.length - 1) {
        setTimeout(() => stream.end(), 500);
      }
    }, (i + 1) * 1000);
  });
}

// === 运行全部测试 ===
async function main() {
  console.log('🚀 开始 gRPC 客户端测试\n');

  // 1. 创建订单
  const orderId = await testCreateOrder();
  console.log('');

  // 2. 查询订单
  // ... (GetOrder 省略，模式相同)

  // 3. 批量上传商品
  await testUploadItems(orderId);
  console.log('');

  // 4. 流式列出订单
  testListOrders('u-001');

  // 5. 实时聊天
  setTimeout(() => testOrderChat(orderId), 2000);
}

main().catch(console.error);

npx tsx src/client/index.ts

🚀 开始 gRPC 客户端测试

[gRPC] --> /order.v1.OrderService/CreateOrder
✅ 订单创建成功: 550e8400-e29b-41d4-a716-446655440000, 金额: ¥798
[gRPC] <-- /order.v1.OrderService/CreateOrder OK (45ms)

[gRPC] --> /order.v1.OrderService/UploadOrderItems
  ↑ 发送商品: 显示器
  ↑ 发送商品: 鼠标
  ↑ 发送商品: 鼠标垫
✅ 上传完成: 5 件, 总计 ¥4943
[gRPC] <-- /order.v1.OrderService/UploadOrderItems OK (120ms)

📦 开始接收订单流...
  - 订单 550e8400...: ¥4943
📦 订单流结束

  → customer: 请问什么时候发货？
  ← system: [已收到] customer: 请问什么时候发货？
  → customer: 能改地址吗？
  ← system: [已收到] customer: 能改地址吗？

请为 gRPC 服务添加：
1. grpc.health.v1 健康检查服务
2. gRPC Server Reflection（服务反射）
使用 @grpc/grpc-js 和 grpc-reflection 包

import { HealthImplementation } from 'grpc-health-check';
import { grpc } from '@grpc/grpc-js';

// 健康状态
const healthImpl = new HealthImplementation({
  'order.v1.OrderService': 'SERVING',
});
healthImpl.addToServer(server);

// 服务反射（让 grpcurl 能自动发现服务）
import { ReflectionService } from 'grpc-reflection';
const reflection = new ReflectionService({
  protoPath: './proto/order.proto',
  serviceName: 'order.v1.OrderService',
});
reflection.addToServer(server);

npm install grpc-health-check grpc-reflection

grpcurl -plaintext localhost:50051 list
# 输出: order.v1.OrderService
#       grpc.health.v1.Health

grpcurl -plaintext localhost:50051 grpc.health.v1.Health/Check
# 输出: { "status": "SERVING" }

请编写 tests/benchmark.ts，使用以下方案对 gRPC 服务进行负载测试：
1. 并发 100 个 Unary CreateOrder 请求，测量 P50/P95/P99 延迟
2. 测试 Server Streaming 吞吐量（每秒推送多少条订单）
3. 测试 Bidi Streaming 的消息往返延迟
4. 输出结果到控制台，包含柱状图 ASCII 展示

import * as grpc from '@grpc/grpc-js';
import { OrderServiceClient } from '../generated/OrderService_grpc_pb';
import { CreateOrderRequest, OrderItem } from '../generated/order_pb';

const CONCURRENT = 100;
const client = new OrderServiceClient(
  'localhost:50051',
  grpc.credentials.createInsecure()
);
const metadata = new grpc.Metadata();
metadata.set('authorization', 'Bearer valid-secret-token');

function createOrder(): Promise<number> {
  return new Promise((resolve, reject) => {
    const item = new OrderItem();
    item.setProductId('p-bench');
    item.setProductName('基准测试商品');
    item.setQuantity(1);
    item.setUnitPrice(100);

    const request = new CreateOrderRequest();
    request.setUserId('bench-user');
    request.setItemsList([item]);
    request.setShippingAddress('测试地址');

    const start = performance.now();
    client.createOrder(request, metadata, (err) => {
      if (err) return reject(err);
      resolve(performance.now() - start);
    });
  });
}

async function runBenchmark() {
  console.log(`⚡ 启动负载测试: ${CONCURRENT} 并发请求\n`);

  const latencies: number[] = [];
  const promises = Array.from({ length: CONCURRENT }, () =>
    createOrder().then((lat) => latencies.push(lat))
  );

  await Promise.all(promises);

  latencies.sort((a, b) => a - b);
  const p50 = latencies[Math.floor(CONCURRENT * 0.5)];
  const p95 = latencies[Math.floor(CONCURRENT * 0.95)];
  const p99 = latencies[Math.floor(CONCURRENT * 0.99)];
  const avg = latencies.reduce((s, v) => s + v, 0) / latencies.length;
  const rps = (CONCURRENT / (latencies[latencies.length - 1] / 1000)).toFixed(0);

  console.log('📊 测试结果:');
  console.log(`  平均延迟: ${avg.toFixed(1)}ms`);
  console.log(`  P50: ${p50.toFixed(1)}ms`);
  console.log(`  P95: ${p95.toFixed(1)}ms`);
  console.log(`  P99: ${p99.toFixed(1)}ms`);
  console.log(`  吞吐量: ~${rps} RPS`);
  console.log('');
  console.log('延迟分布:');
  [0, 10, 20, 30, 50, 80, 100, 150, 200].forEach((bucket) => {
    const count = latencies.filter((l) => l >= bucket && l < bucket + 10).length;
    const bar = '█'.repeat(count);
    console.log(`  ${String(bucket).padStart(3)}ms | ${bar} (${count})`);
  });
}

runBenchmark().catch(console.error);

⚡ 启动负载测试: 100 并发请求

📊 测试结果:
  平均延迟: 28.5ms
  P50: 22.1ms
  P95: 65.3ms
  P99: 89.7ms
  吞吐量: ~3500 RPS

延迟分布:
    0ms | ████████ (8)
   10ms | ████████████████████████████ (28)
   20ms | ████████████████████████████████ (32)
   30ms | ████████████ (12)
   50ms | ████████████ (12)
   80ms | ████ (4)
  100ms | ████ (4)

graph TB
    subgraph "客户端层"
        WEB[Web 前端<br/>gRPC-Web]
        MOBILE[移动端<br/>gRPC Native]
        CLI[gRPCurl<br/>调试工具]
    end

    subgraph "API Gateway"
        GW[Envoy/Nginx<br/>gRPC-HTTP 转码]
    end

    subgraph "gRPC 服务集群"
        OS[OrderService<br/>订单服务 :50051]
        PS[ProductService<br/>商品服务 :50052]
        US[UserService<br/>用户服务 :50053]
    end

    subgraph "横切关注点"
        INT[拦截器链<br/>日志→认证→限流→错误]
        HC[健康检查<br/>grpc.health.v1]
        RF[服务反射<br/>Server Reflection]
    end

    subgraph "数据层"
        REDIS[(Redis<br/>订单缓存)]
        PG[(PostgreSQL<br/>持久化)]
    end

    WEB -->|HTTP/2| GW
    MOBILE -->|gRPC| GW
    CLI -->|gRPC| OS
    GW -->|gRPC| OS
    GW -->|gRPC| PS
    GW -->|gRPC| US

    OS --> INT
    PS --> INT
    US --> INT

    OS --> HC
    OS --> RF

    OS --> REDIS
    OS --> PG
    PS --> PG
    US --> PG

    style OS fill:#4A90D9,color:#fff
    style INT fill:#F5A623,color:#fff
    style GW fill:#7B68EE,color:#fff

sequenceDiagram
    participant C as Client
    participant I as Interceptor
    participant S as OrderService
    participant DB as PostgreSQL
    participant R as Redis

    Note over C,S: Unary RPC: CreateOrder
    C->>I: 1. 发送 CreateOrderRequest
    I->>I: 2. 日志记录 + Token认证
    I->>S: 3. 转发请求
    S->>DB: 4. 插入订单
    DB-->>S: 5. 返回 order_id
    S->>R: 6. 缓存订单
    S-->>I: 7. 返回 CreateOrderResponse
    I-->>C: 8. 响应 + 耗时日志

    Note over C,S: Server Streaming: ListOrders
    C->>I: 9. 发送 ListOrdersRequest
    I->>S: 10. 转发请求
    loop 每条订单
        S-->>I: 11. call.write(order)
        I-->>C: 12. 流式推送
    end
    S-->>C: 13. call.end()

    Note over C,S: Bidi Streaming: OrderChat
    loop 实时交互
        C->>S: 14. call.write(message)
        S-->>C: 15. call.write(reply)
    end

# OrderService - gRPC 微服务

## 技术栈
- Node.js 20 + TypeScript
- gRPC (@grpc/grpc-js)
- Protocol Buffers 3
- Vitest 测试框架

## 项目规范
- Proto 文件放在 proto/ 目录，使用 proto3 语法
- 所有 RPC 方法必须有错误处理和参数校验
- 拦截器放在 src/server/interceptors.ts
- 字段编号不可复用，删除用 reserved
- 生成的代码不手动编辑

## 常用命令
- 代码生成: `./scripts/generate.sh`
- 启动服务: `npx tsx src/server/index.ts`
- 运行客户端: `npx tsx src/client/index.ts`
- 负载测试: `npx tsx tests/benchmark.ts`
- 单元测试: `npx vitest run`

请基于 proto/order.proto，用 Go 语言实现一个 gRPC 客户端：
1. 使用 google.golang.org/grpc 和 google.golang.org/protobuf
2. 连接到 localhost:50051
3. 调用 CreateOrder 方法
4. 添加 5s 超时和重试逻辑