feat: add tbank tokens

2026-08-13 19:31:43 +03:00
parent 4cbfc2621c
commit 31cbba8654
25 changed files with 3592 additions and 24 deletions
@@ -0,0 +1,181 @@
+import { Injectable, Logger, ServiceUnavailableException } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import {
+  ChannelCredentials,
+  Client,
+  ClientUnaryCall,
+  loadPackageDefinition,
+  Metadata,
+  ServiceError,
+  status,
+} from '@grpc/grpc-js';
+import { loadSync } from '@grpc/proto-loader';
+import { existsSync, readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { Environment } from '../config/env.validation';
+
+type GetAccounts = (
+  request: { status: string },
+  metadata: Metadata,
+  options: { deadline: Date },
+  callback: (error: ServiceError | null) => void,
+) => ClientUnaryCall;
+type UsersClient = Client & { getAccounts: GetAccounts };
+type TbankNamespace = {
+  tinkoff: {
+    public: {
+      invest: {
+        api: {
+          contract: { v1: { UsersService: unknown } };
+        };
+      };
+    };
+  };
+};
+
+class RateLimiter {
+  private readonly waiting: (() => void)[] = [];
+  private starts = 0;
+  private windowStartedAt = Date.now();
+
+  constructor(private readonly perSecond: number) {}
+
+  get size(): number {
+    return this.waiting.length;
+  }
+
+  async run<T>(operation: () => Promise<T>): Promise<T> {
+    await new Promise<void>((resolve) => {
+      this.waiting.push(resolve);
+      this.drain();
+    });
+    try {
+      return await operation();
+    } finally {
+      this.drain();
+    }
+  }
+
+  private drain(): void {
+    const now = Date.now();
+    if (now - this.windowStartedAt >= 1000) {
+      this.windowStartedAt = now;
+      this.starts = 0;
+    }
+    while (this.starts < this.perSecond && this.waiting.length) {
+      this.starts += 1;
+      this.waiting.shift()?.();
+    }
+    if (this.waiting.length && this.starts === this.perSecond) {
+      setTimeout(
+        () => this.drain(),
+        Math.max(1, 1000 - (now - this.windowStartedAt)),
+      );
+    }
+  }
+}
+
+@Injectable()
+export class TbankClient {
+  private readonly logger = new Logger(TbankClient.name);
+  private readonly queue: RateLimiter;
+  private readonly client: UsersClient;
+
+  constructor(config: ConfigService<Environment, true>) {
+    const sourceProtoRoot = join(process.cwd(), 'src/tbank/proto/contracts');
+    const builtProtoRoot = join(process.cwd(), 'dist/tbank/proto/contracts');
+    const protoRoot = existsSync(sourceProtoRoot)
+      ? sourceProtoRoot
+      : builtProtoRoot;
+    const definition = loadSync(join(protoRoot, 'users.proto'), {
+      includeDirs: [protoRoot],
+      keepCase: false,
+      longs: String,
+      enums: String,
+      defaults: true,
+      oneofs: true,
+    });
+    const packageDefinition = loadPackageDefinition(
+      definition,
+    ) as unknown as TbankNamespace;
+    const Service = packageDefinition.tinkoff.public.invest.api.contract.v1
+      .UsersService as new (
+      address: string,
+      credentials: ChannelCredentials,
+    ) => UsersClient;
+    const caPath = config.get<string | undefined>('TBANK_CA_CERT_PATH');
+    const credentials = caPath
+      ? ChannelCredentials.createSsl(readFileSync(caPath))
+      : ChannelCredentials.createSsl();
+    this.client = new Service(
+      config.getOrThrow('TBANK_GRPC_ENDPOINT'),
+      credentials,
+    );
+    this.queue = new RateLimiter(
+      config.getOrThrow('TBANK_RATE_LIMIT_PER_SECOND'),
+    );
+    this.logger.log({
+      event: 'client_configured',
+      endpoint: config.getOrThrow('TBANK_GRPC_ENDPOINT'),
+      rateLimitPerSecond: config.getOrThrow('TBANK_RATE_LIMIT_PER_SECOND'),
+      customCa: Boolean(caPath),
+    });
+  }
+
+  async validateToken(token: string): Promise<void> {
+    await this.call(token, true);
+  }
+
+  private async call(token: string, retry: boolean): Promise<void> {
+    if (this.queue.size >= 50) {
+      this.logger.warn({ event: 'queue_full', waiting: this.queue.size });
+      throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
+    }
+    try {
+      this.logger.debug({ event: 'get_accounts_started', retry: !retry });
+      await this.queue.run(() => this.request(token));
+      this.logger.debug({ event: 'get_accounts_succeeded', retry: !retry });
+    } catch (error) {
+      const code = (error as ServiceError).code;
+      this.logger.warn({
+        event: 'get_accounts_failed',
+        grpcCode: code,
+        retry: !retry,
+        retryable: [
+          status.UNAVAILABLE,
+          status.DEADLINE_EXCEEDED,
+          status.RESOURCE_EXHAUSTED,
+        ].includes(code),
+      });
+      if (
+        retry &&
+        [
+          status.UNAVAILABLE,
+          status.DEADLINE_EXCEEDED,
+          status.RESOURCE_EXHAUSTED,
+        ].includes(code)
+      ) {
+        this.logger.log({ event: 'get_accounts_retry_scheduled', delayMs: 1000 });
+        await new Promise((resolve) => setTimeout(resolve, 1000));
+        return this.call(token, false);
+      }
+      if ([status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(code)) {
+        throw new Error('INVALID_TBANK_TOKEN');
+      }
+      throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
+    }
+  }
+
+  private request(token: string): Promise<void> {
+    return new Promise((resolve, reject) => {
+      const metadata = new Metadata();
+      metadata.set('Authorization', `Bearer ${token}`);
+      this.client.getAccounts(
+        { status: 'ACCOUNT_STATUS_OPEN' },
+        metadata,
+        { deadline: new Date(Date.now() + 10_000) },
+        (error) => (error ? reject(error) : resolve()),
+      );
+    });
+  }
+}