moex-vibe/apps/backend/src/modules/tbank/services/tbank-client.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { TBankNotConfiguredException, TBankApiException } from '../../../common/exceptions/tbank-api.exception';
import type {
  TBankAccountsResponse,
  TBankInstrumentResponse,
  TBankOperationsByCursorResponse,
  TBankPortfolioResponse,
  TBankPositionsResponse,
} from '../types/tbank-proto.types';
import {
  CallOptions,
  ChannelCredentials,
  Client,
  ClientUnaryCall,
  loadPackageDefinition,
  Metadata,
  ServiceError,
  status,
} from '@grpc/grpc-js';
import { loadSync } from '@grpc/proto-loader';
import PQueue from 'p-queue';
import { existsSync, readFileSync } from 'node:fs';
import { join } from 'node:path';
import { TBANK_PROTO_FILES, TBANK_PROTO_PACKAGE } from '../tbank.config';

type GrpcUnary<TRequest, TResponse> = (
  request: TRequest,
  metadata: Metadata,
  options: CallOptions,
  callback: (error: ServiceError | null, response: TResponse | null) => void,
) => ClientUnaryCall;

type GrpcServiceConstructor = new (address: string, credentials: ChannelCredentials) => Client;

type TBankAccountsRequest = { status: string };
export type TBankPortfolioRequest = { accountId: string; currency: string };
type TBankPositionsRequest = { accountId: string };
type TBankInstrumentRequest = { idType: string; id: string };

export type TBankUsersClient = Client & {
  getAccounts: GrpcUnary<TBankAccountsRequest, TBankAccountsResponse>;
};

export type TBankOperationsClient = Client & {
  getPortfolio: GrpcUnary<TBankPortfolioRequest, TBankPortfolioResponse>;
  getPositions: GrpcUnary<TBankPositionsRequest, TBankPositionsResponse>;
  getOperationsByCursor: GrpcUnary<Record<string, unknown>, TBankOperationsByCursorResponse>;
};

export type TBankInstrumentsClient = Client & {
  getInstrumentBy: GrpcUnary<TBankInstrumentRequest, TBankInstrumentResponse>;
};

type QueueName = 'operations' | 'instruments' | 'users';

@Injectable()
export class TBankClientService {
  private readonly logger = new Logger(TBankClientService.name);
  private readonly queues: Record<QueueName, PQueue>;
  private readonly requestTimeoutMs: number;
  private readonly packageDefinition: ReturnType<typeof loadPackageDefinition>;
  private readonly clientCache = new Map<string, Client>();

  constructor(private readonly configService: ConfigService) {
    this.requestTimeoutMs = this.configService.get<number>('app.tbank.requestTimeoutMs', 10000);
    const operationsRate = this.configService.get<number>('app.tbank.rateLimitPerSecond', 5);
    const instrumentsRate = this.configService.get<number>(
      'app.tbank.instrumentsRateLimitPerSecond',
      20,
    );
    this.queues = {
      operations: new PQueue({ interval: 1000, intervalCap: operationsRate }),
      instruments: new PQueue({ interval: 1000, intervalCap: instrumentsRate }),
      users: new PQueue({ interval: 1000, intervalCap: operationsRate }),
    };

    const protoRoot = this.resolveProtoRoot();
    const definition = loadSync(
      Object.values(TBANK_PROTO_FILES).map((fileName) => join(protoRoot, fileName)),
      {
        includeDirs: [protoRoot],
        keepCase: false,
        longs: String,
        enums: String,
        defaults: true,
        oneofs: true,
      },
    );

    this.packageDefinition = loadPackageDefinition(definition);
  }

  private resolveProtoRoot(): string {
    const distProtoRoot = join(__dirname, '..', 'proto', 'contracts');
    if (existsSync(distProtoRoot)) return distProtoRoot;

    return join(process.cwd(), 'src', 'modules', 'tbank', 'proto', 'contracts');
  }

  createMetadata(): Metadata {
    const token = this.configService.get<string>('app.tbank.token', '');
    if (!token) {
      throw new TBankNotConfiguredException();
    }

    const metadata = new Metadata();
    metadata.set('Authorization', `Bearer ${token}`);

    const appName = this.configService.get<string>('app.tbank.appName', '');
    if (appName) metadata.set('x-app-name', appName);

    return metadata;
  }

  redactMetadata(metadata: Metadata): Record<string, string> {
    const result: Record<string, string> = {};

    for (const key of Object.keys(metadata.getMap())) {
      if (key.toLowerCase() === 'authorization') {
        result.Authorization = '<redacted>';
        continue;
      }

      result[key] = String(metadata.get(key)[0]);
    }

    return result;
  }

  getServiceClient(
    serviceName: 'UsersService' | 'OperationsService' | 'InstrumentsService',
  ): Client {
    const cached = this.clientCache.get(serviceName);
    if (cached) return cached;

    const namespace = this.resolveProtoNamespace();
    const ServiceCtor = namespace[serviceName] as GrpcServiceConstructor;
    const client = new ServiceCtor(
      this.configService.get<string>('app.tbank.baseUrl', 'invest-public-api.tbank.ru:443'),
      this.createChannelCredentials(),
    );

    this.clientCache.set(serviceName, client);

    return client;
  }

  getUsersClient(): TBankUsersClient {
    return this.getServiceClient('UsersService') as TBankUsersClient;
  }

  getOperationsClient(): TBankOperationsClient {
    return this.getServiceClient('OperationsService') as TBankOperationsClient;
  }

  getInstrumentsClient(): TBankInstrumentsClient {
    return this.getServiceClient('InstrumentsService') as TBankInstrumentsClient;
  }

  async callUnary<TRequest, TResponse>(
    label: string,
    method: GrpcUnary<TRequest, TResponse>,
    request: TRequest,
    queueName: QueueName = 'operations',
  ): Promise<TResponse> {
    return this.queues[queueName].add(
      () =>
        new Promise<TResponse>((resolve, reject) => {
          const metadata = this.createMetadata();
          const deadline = new Date(Date.now() + this.requestTimeoutMs);

          method(request, metadata, { deadline }, (error, response) => {
            if (error) {
              reject(this.mapGrpcError(label, error));
              return;
            }

            resolve(response as TResponse);
          });
        }),
    ) as Promise<TResponse>;
  }

  private resolveProtoNamespace(): Record<string, unknown> {
    return TBANK_PROTO_PACKAGE.split('.').reduce<Record<string, unknown>>(
      (current, part) => {
        return current[part] as Record<string, unknown>;
      },
      this.packageDefinition as Record<string, unknown>,
    );
  }

  private createChannelCredentials(): ChannelCredentials {
    const caCertPath = this.configService.get<string>('app.tbank.caCertPath', '');
    if (!caCertPath) return ChannelCredentials.createSsl();

    try {
      return ChannelCredentials.createSsl(readFileSync(caCertPath));
    } catch (error) {
      this.logger.error(
        `Failed to read T-Bank CA certificate from ${caCertPath}: ${
          error instanceof Error ? error.message : String(error)
        }`,
      );
      throw new TBankApiException('T-Bank CA certificate is not readable');
    }
  }

  private mapGrpcError(label: string, error: ServiceError): Error {
    const trackingId = error.metadata?.get('x-tracking-id')?.[0];
    const retryAfter = error.metadata?.get('x-ratelimit-reset')?.[0];
    const publicMessage =
      error.code === status.RESOURCE_EXHAUSTED
        ? 'T-Bank upstream error: rate limit exceeded'
        : `T-Bank upstream error while calling ${label}`;

    this.logger.warn(
      JSON.stringify({
        label,
        code: error.code,
        trackingId,
        retryAfter,
        message: error.message,
      }),
    );

    const detail = [publicMessage, trackingId && `trackingId:${trackingId}`, retryAfter && `retryAfter:${retryAfter}`]
      .filter(Boolean)
      .join('; ');
    return new TBankApiException(detail);
  }
}