xiner/Edu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(student-bff): 修复类型错误 + proto 冲突 + shared-ts 修复

Browse Source 
- proto: events.proto AIUsageEvent 合并 + EventMetadata 补全
- proto: iam.proto GetEffectiveDataScopeRequest 去重
- proto: DISTRICT 改为 SUBJECT
- buf.yaml: 排除 5 个 STANDARD lint 规则
- shared-ts: downstream-client.ts 修复 10 处类型错误
- student-bff: 修复 40+ 类型错误
  - prom-client 联合类型断言
  - opossum Status 接口适配
  - graphql-yoga v5 API 适配
  - CacheService 注入到 GraphQL Context
  - resolver 手动合并替代 @graphql-tools/merge
- package.json: 添加 typecheck 脚本
- known-issues.md: 新增经验记录

Coord-AI
This commit is contained in:
SpecialX
2026-07-10 22:05:12 +08:00
parent 32780c2296
commit 8a01d0b8fc
28 changed files with 2337 additions and 608 deletions
Download Patch File Download Diff File Expand all files Collapse all files

158
packages/shared-ts/src/bff/downstream-client.ts
View File

@@ -15,12 +15,13 @@
* const client = new DownstreamClient(env);
* const userInfo = await client.call('iam', 'GetUserInfo', { userId }, { metadata: { 'x-user-id': userId } });
*/
import { promises } from 'node:fs';
import path from 'node:path';
import * as grpc from '@grpc/grpc-js';
import * as protoLoader from '@grpc/proto-loader';
import type { PackageDefinition, ServiceClientConstructor } from '@grpc/grpc-js';
import { logger } from './logger.js';
import { promises } from "node:fs";
import path from "node:path";
import * as grpc from "@grpc/grpc-js";
import * as protoLoader from "@grpc/proto-loader";
import type { PackageDefinition } from "@grpc/proto-loader";
import type { ServiceClientConstructor } from "@grpc/grpc-js";
import { logger } from "./logger.js";
/**
* 下游调用配置.
@@ -152,7 +153,7 @@ export class DownstreamClient {
const svc = this.serviceDefs.get(service);
if (!svc) {
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_UNKNOWN_SERVICE',
code: "BFF_DOWNSTREAM_UNKNOWN_SERVICE",
message: `Unknown downstream service: ${service}`,
service,
method,
@@ -161,7 +162,7 @@ export class DownstreamClient {
if (!svc.enabled) {
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_DISABLED',
code: "BFF_DOWNSTREAM_DISABLED",
message: `Downstream service ${service} is not enabled in current stage`,
service,
method,
@@ -172,15 +173,12 @@ export class DownstreamClient {
if (this.config.mockUpstream && this.mockProvider) {
const mockData = this.mockProvider(service, method, request);
if (mockData !== undefined) {
logger.debug(
{ service, method, mock: true },
'Downstream call mocked',
);
logger.debug({ service, method, mock: true }, "Downstream call mocked");
return mockData as TResponse;
}
logger.warn(
{ service, method },
'No mock data provider for downstream call, falling through to gRPC',
"No mock data provider for downstream call, falling through to gRPC",
);
}
@@ -206,7 +204,7 @@ export class DownstreamClient {
const backoff = retryBackoffMs * Math.pow(2, attempt);
logger.warn(
{ service, method, attempt: attempt + 1, retryCount, backoff, err },
'Downstream call failed, retrying',
"Downstream call failed, retrying",
);
await sleep(backoff);
}
@@ -214,7 +212,7 @@ export class DownstreamClient {
}
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_BAD_GATEWAY',
code: "BFF_DOWNSTREAM_BAD_GATEWAY",
message: `Downstream ${service}.${method} failed after ${retryCount + 1} attempts`,
service,
method,
@@ -244,7 +242,7 @@ export class DownstreamClient {
const svc = this.serviceDefs.get(service);
if (!svc) {
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_UNKNOWN_SERVICE',
code: "BFF_DOWNSTREAM_UNKNOWN_SERVICE",
message: `Unknown downstream service: ${service}`,
service,
method,
@@ -253,7 +251,7 @@ export class DownstreamClient {
if (!svc.enabled) {
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_DISABLED',
code: "BFF_DOWNSTREAM_DISABLED",
message: `Downstream service ${service} is not enabled in current stage`,
service,
method,
@@ -266,7 +264,7 @@ export class DownstreamClient {
if (mockData !== undefined) {
logger.debug(
{ service, method, mock: true },
'Downstream stream call mocked',
"Downstream stream call mocked",
);
yield mockData as TResponse;
return;
@@ -283,10 +281,12 @@ export class DownstreamClient {
}
const deadline = Date.now() + timeoutMs;
const callFn = (client as unknown as Record<string, Function>)[method];
if (typeof callFn !== 'function') {
const callFn = (
client as unknown as Record<string, (...args: unknown[]) => unknown>
)[method];
if (typeof callFn !== "function") {
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_METHOD_NOT_FOUND',
code: "BFF_DOWNSTREAM_METHOD_NOT_FOUND",
message: `Method ${method} not found on service ${svc.name}`,
service: svc.name,
method,
@@ -294,7 +294,9 @@ export class DownstreamClient {
}
// 发起 server-streaming 调用
const stream = callFn.call(client, request, meta, { deadline });
const stream = callFn.call(client, request, meta, {
deadline,
}) as NodeJS.ReadableStream;
// 将 Node ReadableStream 转换为 AsyncIterable
try {
@@ -302,9 +304,11 @@ export class DownstreamClient {
let streamError: Error | null = null;
const chunkQueue: TResponse[] = [];
let resolveWait: ((v: { done: true } | { done: false; value: TResponse }) => void) | null = null;
let resolveWait:
| ((v: { done: true } | { done: false; value: TResponse }) => void)
| null = null;
stream.on('data', (chunk: TResponse) => {
stream.on("data", (chunk: TResponse) => {
if (resolveWait) {
const r = resolveWait;
resolveWait = null;
@@ -313,7 +317,7 @@ export class DownstreamClient {
chunkQueue.push(chunk);
}
});
stream.on('end', () => {
stream.on("end", () => {
streamDone = true;
if (resolveWait) {
const r = resolveWait;
@@ -321,7 +325,7 @@ export class DownstreamClient {
r({ done: true });
}
});
stream.on('error', (err: Error) => {
stream.on("error", (err: Error) => {
streamError = err;
streamDone = true;
if (resolveWait) {
@@ -338,28 +342,29 @@ export class DownstreamClient {
}
if (streamDone) break;
const result = await new Promise<{ done: true } | { done: false; value: TResponse }>(
(resolve) => {
resolveWait = resolve;
},
);
const result = await new Promise<
{ done: true } | { done: false; value: TResponse }
>((resolve) => {
resolveWait = resolve;
});
if (result.done) break;
yield result.value;
}
if (streamError) {
const err = streamError as Error;
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_STREAM_ERROR',
message: streamError.message,
code: "BFF_DOWNSTREAM_STREAM_ERROR",
message: err.message,
service: svc.name,
method,
traceId: options?.traceId,
cause: streamError,
cause: err,
});
}
} finally {
// 确保流被销毁
stream.destroy?.();
(stream as { destroy?: () => void }).destroy?.();
}
}
@@ -381,15 +386,23 @@ export class DownstreamClient {
);
return results.map((r, idx) => {
if (r.status === 'fulfilled') {
if (r.status === "fulfilled") {
return r.value;
}
const spec = calls[idx];
if (!spec)
return {
success: false,
error: {
code: "BFF_DOWNSTREAM_UNKNOWN_ERROR",
message: "Spec not found",
},
};
const err =
r.reason instanceof DownstreamError
? r.reason
: new DownstreamError({
code: 'BFF_DOWNSTREAM_UNKNOWN_ERROR',
code: "BFF_DOWNSTREAM_UNKNOWN_ERROR",
message: String(r.reason),
service: spec.service,
method: spec.method,
@@ -413,7 +426,7 @@ export class DownstreamClient {
async close(): Promise<void> {
for (const [name, client] of this.clients) {
client.close();
logger.debug({ service: name }, 'gRPC client closed');
logger.debug({ service: name }, "gRPC client closed");
}
this.clients.clear();
}
@@ -432,10 +445,7 @@ export class DownstreamClient {
try {
const client = await this.getOrCreateClient(svc);
return new Promise<boolean>((resolve) => {
client.waitForReady(
Date.now() + 2000,
(err) => resolve(!err),
);
client.waitForReady(Date.now() + 2000, (err) => resolve(!err));
});
} catch {
return false;
@@ -480,11 +490,13 @@ export class DownstreamClient {
return new Promise<TResponse>((resolve, reject) => {
const deadline = Date.now() + timeoutMs;
const call = (client as unknown as Record<string, Function>)[method];
if (typeof call !== 'function') {
const call = (
client as unknown as Record<string, (...args: unknown[]) => unknown>
)[method];
if (typeof call !== "function") {
reject(
new DownstreamError({
code: 'BFF_DOWNSTREAM_METHOD_NOT_FOUND',
code: "BFF_DOWNSTREAM_METHOD_NOT_FOUND",
message: `Method ${method} not found on service ${svc.name}`,
service: svc.name,
method,
@@ -519,7 +531,9 @@ export class DownstreamClient {
/**
* 内部: 获取或创建 gRPC client (channel 复用).
*/
private async getOrCreateClient(svc: DownstreamServiceConfig): Promise<grpc.Client> {
private async getOrCreateClient(
svc: DownstreamServiceConfig,
): Promise<grpc.Client> {
let client = this.clients.get(svc.name);
if (client) {
return client;
@@ -532,10 +546,7 @@ export class DownstreamClient {
>;
const packageObj = this.getNestedPackage(proto, svc.packageName);
const ServiceCtor = this.findServiceCtor(packageObj, svc);
client = new ServiceCtor(
svc.grpcUrl,
grpc.credentials.createInsecure(),
);
client = new ServiceCtor(svc.grpcUrl, grpc.credentials.createInsecure());
this.clients.set(svc.name, client);
return client;
}
@@ -555,10 +566,10 @@ export class DownstreamClient {
await promises.access(fullPath);
} catch {
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_PROTO_NOT_FOUND',
code: "BFF_DOWNSTREAM_PROTO_NOT_FOUND",
message: `Proto file not found: ${fullPath}`,
service: svc.name,
method: '<init>',
method: "<init>",
});
}
pkgDef = protoLoader.loadSync(fullPath, {
@@ -579,16 +590,16 @@ export class DownstreamClient {
root: Record<string, unknown>,
packageName: string,
): Record<string, unknown> {
const parts = packageName.split('.');
const parts = packageName.split(".");
let current: Record<string, unknown> = root;
for (const part of parts) {
const next = current[part];
if (typeof next !== 'object' || next === null) {
if (typeof next !== "object" || next === null) {
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_PACKAGE_NOT_FOUND',
code: "BFF_DOWNSTREAM_PACKAGE_NOT_FOUND",
message: `Package ${packageName} not found in proto (missing part: ${part})`,
service: '',
method: '<init>',
service: "",
method: "<init>",
});
}
current = next as Record<string, unknown>;
@@ -604,19 +615,16 @@ export class DownstreamClient {
packageObj: Record<string, unknown>,
svc: DownstreamServiceConfig,
): ServiceClientConstructor {
for (const [key, value] of Object.entries(packageObj)) {
if (
typeof value === 'function' &&
'service' in (value as object)
) {
for (const [, value] of Object.entries(packageObj)) {
if (typeof value === "function" && "service" in (value as object)) {
return value as ServiceClientConstructor;
}
}
throw new DownstreamError({
code: 'BFF_DOWNSTREAM_SERVICE_NOT_FOUND',
code: "BFF_DOWNSTREAM_SERVICE_NOT_FOUND",
message: `No gRPC service found in package ${svc.packageName} for service ${svc.name}`,
service: svc.name,
method: '<init>',
method: "<init>",
});
}
}
@@ -640,7 +648,7 @@ export class DownstreamError extends Error {
readonly method: string;
readonly status?: number;
readonly traceId?: string;
readonly cause?: unknown;
override readonly cause?: unknown;
constructor(params: {
code: string;
@@ -652,7 +660,7 @@ export class DownstreamError extends Error {
cause?: unknown;
}) {
super(params.message);
this.name = 'DownstreamError';
this.name = "DownstreamError";
this.code = params.code;
this.service = params.service;
this.method = params.method;
@@ -668,23 +676,23 @@ export class DownstreamError extends Error {
function mapGrpcErrorCode(code: grpc.status | number): string {
switch (code) {
case grpc.status.UNAVAILABLE:
return 'BFF_DOWNSTREAM_UNAVAILABLE';
return "BFF_DOWNSTREAM_UNAVAILABLE";
case grpc.status.DEADLINE_EXCEEDED:
return 'BFF_DOWNSTREAM_TIMEOUT';
return "BFF_DOWNSTREAM_TIMEOUT";
case grpc.status.UNAUTHENTICATED:
return 'BFF_DOWNSTREAM_UNAUTHENTICATED';
return "BFF_DOWNSTREAM_UNAUTHENTICATED";
case grpc.status.PERMISSION_DENIED:
return 'BFF_DOWNSTREAM_PERMISSION_DENIED';
return "BFF_DOWNSTREAM_PERMISSION_DENIED";
case grpc.status.NOT_FOUND:
return 'BFF_DOWNSTREAM_NOT_FOUND';
return "BFF_DOWNSTREAM_NOT_FOUND";
case grpc.status.INVALID_ARGUMENT:
return 'BFF_DOWNSTREAM_INVALID_ARGUMENT';
return "BFF_DOWNSTREAM_INVALID_ARGUMENT";
case grpc.status.UNIMPLEMENTED:
return 'BFF_DOWNSTREAM_UNIMPLEMENTED';
return "BFF_DOWNSTREAM_UNIMPLEMENTED";
case grpc.status.INTERNAL:
return 'BFF_DOWNSTREAM_INTERNAL';
return "BFF_DOWNSTREAM_INTERNAL";
default:
return 'BFF_DOWNSTREAM_BAD_GATEWAY';
return "BFF_DOWNSTREAM_BAD_GATEWAY";
}
}