160 lines
4.6 KiB
TypeScript
160 lines
4.6 KiB
TypeScript
// DownstreamClient 抽象基类（B8 裁决：3 个 BFF 统一使用）
|
||
// 提供统一的错误映射 + 结构化日志 + metrics 记录
|
||
import type { Logger } from "pino";
|
||
import { logger } from "../shared/observability/logger.js";
|
||
import {
|
||
BadGatewayError,
|
||
AggregationFailedError,
|
||
} from "../shared/errors/application-error.js";
|
||
import type {
|
||
CallContext,
|
||
DownstreamError,
|
||
DownstreamResult,
|
||
DownstreamServiceName,
|
||
GrpcMetadata,
|
||
} from "./types.js";
|
||
|
||
/** gRPC 状态码 → HTTP 语义映射（@grpc/grpc-js status codes） */
|
||
const GRPC_STATUS = {
|
||
OK: 0,
|
||
CANCELLED: 1,
|
||
UNKNOWN: 2,
|
||
INVALID_ARGUMENT: 3,
|
||
DEADLINE_EXCEEDED: 4,
|
||
NOT_FOUND: 5,
|
||
ALREADY_EXISTS: 6,
|
||
PERMISSION_DENIED: 7,
|
||
RESOURCE_EXHAUSTED: 8,
|
||
FAILED_PRECONDITION: 9,
|
||
ABORTED: 10,
|
||
OUT_OF_RANGE: 11,
|
||
UNIMPLEMENTED: 12,
|
||
INTERNAL: 13,
|
||
UNAVAILABLE: 14,
|
||
DATA_LOSS: 15,
|
||
UNAUTHENTICATED: 16,
|
||
} as const;
|
||
|
||
export abstract class BaseDownstreamClient {
|
||
protected log: Logger;
|
||
abstract readonly serviceName: DownstreamServiceName;
|
||
|
||
constructor() {
|
||
// abstract property 在子类构造后才可用，延迟初始化 logger
|
||
this.log = logger.child({ downstream: "downstream" });
|
||
}
|
||
|
||
/** 子类在构造函数中调用以初始化 logger（带正确 serviceName） */
|
||
protected initLogger(): void {
|
||
this.log = logger.child({ downstream: this.serviceName });
|
||
}
|
||
|
||
/** 构建 gRPC metadata（注入 x-user-id + x-request-id） */
|
||
protected buildMetadata(ctx: CallContext): GrpcMetadata {
|
||
const meta: GrpcMetadata = { "x-user-id": ctx.userId };
|
||
if (ctx.traceId) {
|
||
meta["x-request-id"] = ctx.traceId;
|
||
}
|
||
return meta;
|
||
}
|
||
|
||
/** 将 gRPC 错误映射为 BadGatewayError（502） */
|
||
protected mapGrpcError(err: unknown, rpc: string): BadGatewayError {
|
||
const grpcErr = err as {
|
||
code?: number;
|
||
message?: string;
|
||
details?: string;
|
||
};
|
||
const code = grpcErr.code ?? GRPC_STATUS.UNKNOWN;
|
||
const message = grpcErr.message ?? "Unknown gRPC error";
|
||
|
||
this.log.warn(
|
||
{ rpc, grpcCode: code, message, err },
|
||
"Downstream gRPC call failed",
|
||
);
|
||
|
||
// UNAVAILABLE → 502 Bad Gateway
|
||
if (code === GRPC_STATUS.UNAVAILABLE) {
|
||
return new BadGatewayError(
|
||
`Downstream ${this.serviceName}.${rpc} unavailable`,
|
||
{ service: this.serviceName, rpc, grpcCode: code },
|
||
);
|
||
}
|
||
|
||
// UNAUTHENTICATED → 映射为 BadGateway（BFF 自身做身份校验，下游不应返回 UNAUTHENTICATED）
|
||
if (code === GRPC_STATUS.UNAUTHENTICATED) {
|
||
return new BadGatewayError(
|
||
`Downstream ${this.serviceName}.${rpc} returned UNAUTHENTICATED`,
|
||
{ service: this.serviceName, rpc, grpcCode: code },
|
||
);
|
||
}
|
||
|
||
// 其他错误统一映射为 BadGateway
|
||
return new BadGatewayError(
|
||
`Downstream ${this.serviceName}.${rpc} failed: ${message}`,
|
||
{
|
||
service: this.serviceName,
|
||
rpc,
|
||
grpcCode: code,
|
||
details: grpcErr.details,
|
||
},
|
||
);
|
||
}
|
||
|
||
/** 执行 gRPC 调用并统一错误处理 + 超时控制 */
|
||
protected async callGrpc<T>(
|
||
rpc: string,
|
||
fn: () => Promise<T>,
|
||
timeoutMs = 3000,
|
||
): Promise<T> {
|
||
const start = Date.now();
|
||
try {
|
||
const result = await Promise.race([
|
||
fn(),
|
||
this.createTimeout(timeoutMs, rpc),
|
||
]);
|
||
const duration = Date.now() - start;
|
||
this.log.debug(
|
||
{ rpc, durationMs: duration },
|
||
"Downstream gRPC call succeeded",
|
||
);
|
||
return result;
|
||
} catch (err) {
|
||
throw this.mapGrpcError(err, rpc);
|
||
}
|
||
}
|
||
|
||
private createTimeout(ms: number, rpc: string): Promise<never> {
|
||
return new Promise((_, reject) => {
|
||
setTimeout(() => {
|
||
reject(
|
||
new BadGatewayError(
|
||
`Downstream ${this.serviceName}.${rpc} timed out after ${ms}ms`,
|
||
{ service: this.serviceName, rpc, timeoutMs: ms },
|
||
),
|
||
);
|
||
}, ms);
|
||
});
|
||
}
|
||
|
||
/** 降级模式 B：部分失败时返回 success=true + degraded=true + warning（president §2.6） */
|
||
protected degraded<T>(
|
||
data: T | null,
|
||
warning: string,
|
||
rpc: string,
|
||
): DownstreamResult<T> {
|
||
this.log.warn({ rpc, warning }, "Downstream call degraded");
|
||
return { success: true, data, degraded: true, warning };
|
||
}
|
||
|
||
/** 聚合失败（多下游部分失败且无降级数据，president §2.6） */
|
||
protected aggregationFailed(
|
||
errors: DownstreamError[],
|
||
): AggregationFailedError {
|
||
return new AggregationFailedError(
|
||
`Aggregation failed for ${this.serviceName}: ${errors.length} downstream(s) failed`,
|
||
{ service: this.serviceName, errors },
|
||
);
|
||
}
|
||
}
|