18 代码走读实录

/**
 * 跨 Consul 实例共享的 gRPC 客户端注册表。
 * 每条记录 = (consul 实例, 远端服务) 的一个可订阅客户端槽位；
 * 消费者订阅任意一条，恢复时 updateGrpcClient 会通知所有同服务条目。
 */
const grpcClientRegistry: GrpcClientInterface[] = [];

/** 仅供测试复位注册表 */
export function resetGrpcClientRegistry() {
  grpcClientRegistry.length = 0;
}

async initGrpcClient(options: ConsulServiceOptions, _serviceInfo?: ServiceInfo) {
  const { serviceName, packageName, protoName } = options;
  let item: GrpcClientInterface;
  if (!_serviceInfo) {
    let existing = grpcClientRegistry.find(
      (c) => c.consulName === this.name && c.serviceName === serviceName,
    );
    if (!existing) {
      existing = {
        client: new BehaviorSubject<ClientGrpc | null>(null),
        consulName: this.name,
        consulClient: this.consulClient,
        serviceName,
        consulService: this,
      };
      grpcClientRegistry.push(existing);
    }
    item = existing;
  } else if (_serviceInfo.client) {
    item = _serviceInfo.client;
  }
  // ...
  const serviceInfo = _serviceInfo || (await this.getServiceInfo(serviceName));
  if (serviceInfo) {
    const tlsEnabled = process.env.GRPC_TLS === 'on';
    const client = ClientProxyFactory.create({
      transport: Transport.GRPC,
      options: {
        url: `${serviceInfo.address}:${serviceInfo.port}`,
        package: packageName,
        packageDefinition: loadProto(protoName || packageName),
        ...(tlsEnabled
          ? { credentials: ChannelCredentials.createSsl(loadGrpcTls().rootCerts) }
          : {}),
      },
    });
    item.client.next(client);
  } else {
    // Node 对未处理的 Promise 拒绝会直接崩溃进程，发现失败时必须兜底并重试
    const retryUpdate = () => {
      this.updateGrpcClient(serviceName).catch((err: Error) => {
        this.logger.warn(`updateGrpcClient failed: ${err.message}, retrying in 5s`);
        setTimeout(retryUpdate, 5000);
      });
    };
    retryUpdate();
  }
  return item;
}

async getServiceInfo(serviceName: string, client?: ConsulHttpClient) {
  const consulClient = client || this.consulClient;
  return firstValueFrom(
    from(
      consulClient.health.service({ service: serviceName, passing: true }) as Promise<any[]>,
    ).pipe(
      mergeMap((healthServices) => {
        if (!healthServices.length) {
          return throwError(() => new Error('no service found'));
        }
        const service = healthServices[Math.floor(Math.random() * healthServices.length)];
        return of({
          address: service.Service.Address,
          port: service.Service.Port,
          serviceName,
        });
      }),
      retry({
        count: 3,
        delay: (retryCount) => {
          const delayDuration = Math.pow(2, retryCount) * 1000;
          return timer(delayDuration);
        },
      }),
      catchError((err) => {
        this.logger.warn(`Error fetching service info: ${err.message}`);
        return of(null);
      }),
    ),
  );
}

// 并行在多个consulClient上调用 -> 其中有一个serviceInfo响应回来 -> 中止，返回该serviceInfo
// mergeMap -> 并行  concatMap -> 串行
async retryGetServiceInfo(serviceName: string) {
  const serviceInfoObservable = from(this.grpcClients).pipe(
    mergeMap((client) =>
      from(this.getServiceInfo(serviceName, client.consulClient)).pipe(
        filter((result) => result !== null),
        map((result) => ({ ...result, client })),
      ),
    ),
    take(1),
  );
  const serviceInfo = await firstValueFrom(serviceInfoObservable).catch(
    (err: Error) => {
      // 所有 Consul 实例均未命中时 firstValueFrom 抛 EmptyError，
      // 转成带上下文的常规错误，避免调用侧出现不可诊断的裸异常
      if (err.message === 'no elements in sequence') {
        throw new Error(
          `[${this.name}] no passing instance found for ${serviceName} in any consul`,
        );
      }
      throw err;
    },
  );
  // ...
}

// 更新grpcClient
async updateGrpcClient(serviceName: string) {
  const grpcOptions = Array.isArray(this.services)
    ? this.services.find((opt) => opt.serviceName === serviceName)
    : this.services;
  if (!grpcOptions) {
    throw new Error(`service ${serviceName} not configured on ${this.name}`);
  }
  const serviceInfo = await this.retryGetServiceInfo(serviceName);
  // 多 Consul 实例下订阅方可能挂在任意一条客户端条目上，
  // 恢复时必须通知所有同服务条目，否则订阅在"落选"条目上的消费者永远收不到新 client
  const targets = grpcClientRegistry.filter((c) => c.serviceName === serviceName);
  await Promise.all(
    targets.map((item) =>
      this.initGrpcClient(grpcOptions, { ...serviceInfo, client: item }),
    ),
  );
  return serviceInfo.client;
}

@ApiTags('system')
@ApiBearerAuth()
@UseGuards(AuthGuard('jwt'), PoliciesGuard)
@Controller('system/users')
export class UserController {
  // ...
  @CheckPolicies((ability) => ability.can('system:user:create', 'perm'))
  @Post()
  create(
    @Body() payload: {
      username: string; password: string; nickname?: string;
      phone?: string; email?: string; deptId?: number; roleIds?: number[];
    },
  ) {
    return this.userClient.createUser(payload);
  }

// gateway/src/system/user-client.service.ts
createUser(payload: {
  username: string; password: string; nickname?: string;
  phone?: string; email?: string; deptId?: number; roleIds?: number[];
}) {
  return this.requireClient().createUser(payload);
}

private requireClient(): UserDeptApi {
  if (!this.userService) {
    throw new Error('user service unavailable');
  }
  return this.userService;
}

@GrpcMethod('UserService', 'CreateUser')
async createUser(@Payload() payload: { /* 同上 */ }) {
  try {
    const created = await this.prismaClient.$transaction(async (prisma) => {
      const roleIds = payload.roleIds?.length
        ? payload.roleIds
        : [Number(this.configService.get('ROLE_ID')) || 1];
      const count = await prisma.role.count({ where: { id: { in: roleIds } } });
      if (count !== roleIds.length) {
        throw new RpcException({ code: status.NOT_FOUND, message: '部分角色不存在' });
      }
      return prisma.user.create({
        data: {
          username: payload.username,
          password: await import('argon2').then(a => a.hash(payload.password, { type: a.argon2id })),
          // ...
          UserRole: { createMany: { data: roleIds.map(roleId => ({ roleId })) } },
        },
        include: { UserRole: true, dept: true },
      });
    });
    return { /* UserData 形状 */ };
  } catch (err: any) {
    if (err?.code === 'P2002') {
      throw new RpcException({ code: status.ALREADY_EXISTS, message: '用户名已存在' });
    }
    throw err;
  }
}

/** 请求白名单，放置一些不需要`token`的接口（防止`token`过期后再请求造成的死循环问题） */
const whiteList = ["/auth/signin", "/auth/refresh"];
return whiteList.some(url => config.url.endsWith(url))
  ? config
  : new Promise(resolve => {
      const data = getToken();
      if (data) {
        const now = new Date().getTime();
        const expired = parseInt(data.expires) - now <= 0;
        if (expired) {
          if (!PureHttp.isRefreshing) {
            PureHttp.isRefreshing = true;
            useUserStoreHook()
              .handRefreshToken({ refreshToken: data.refreshToken })
              .then(res => {
                const token = res.data.accessToken;
                config.headers["Authorization"] = formatToken(token);
                PureHttp.requests.forEach(cb => cb(token));
                PureHttp.requests = [];
              })
              .catch(_err => {
                PureHttp.requests = [];
                useUserStoreHook().logOut();
                message("登录已过期，请重新登录", { type: "warning" });
              })
              .finally(() => { PureHttp.isRefreshing = false; });
          }
          resolve(PureHttp.retryOriginalRequest(config));
        } else {
          config.headers["Authorization"] = formatToken(data.accessToken);
          resolve(config);
        }
      } else {
        resolve(config);
      }
    });

(error: PureHttpError) => {
  const $error = error;
  $error.isCancelRequest = Axios.isCancel($error);
  // 网关返回 401：会话过期，清理本地登录态并跳回登录页
  if ($error?.response?.status === 401) {
    clearLoginState();
    if (window.location.pathname !== "/login") {
      window.location.href = "/login";
    }
  }
  return Promise.reject($error);
}

// src/utils/auth.ts
export const hasPerms = (value: string | Array<string>): boolean => {
  if (!value) return false;
  const allPerms = "*:*:*";
  const { permissions } = useUserStoreHook();
  if (!permissions) return false;
  if (permissions.length === 1 && permissions[0] === allPerms) return true;
  const isAuths = isString(value)
    ? permissions.includes(value)
    : isIncludeAllChildren(value, permissions);
  return isAuths ? true : false;
};

// src/directives/perms/index.ts
export const perms: Directive = {
  mounted(el: HTMLElement, binding: DirectiveBinding<string | Array<string>>) {
    const { value } = binding;
    if (value) {
      !hasPerms(value) && el.parentNode?.removeChild(el);
    } else {
      throw new Error("[Directive: perms]: need perms! Like v-perms=\"['btn.add','btn.edit']\"");
    }
  }
};