backend/src/integrations/tbank/instrument-metadata.service.ts

import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';

import { StructuredLogger } from '../../common/logging/structured-logger.service';
import {
  InstrumentMetadataRepository,
  type MetadataIdentity,
  metadataKey,
  type MetadataLookup,
} from './instrument-metadata.repository';
import {
  TbankAccessService,
  type TbankSyncAttempt,
} from './tbank-access.service';
import { isTbankErrorCode, TbankErrorCode } from './tbank-error-code';
import { TbankClient, type TbankInstrument } from './tbank.client';
export { metadataKey } from './instrument-metadata.repository';
export type { MetadataIdentity } from './instrument-metadata.repository';

type Producer = {
  attempt: TbankSyncAttempt;
  identity: MetadataIdentity;
  observers: TbankSyncAttempt[];
};
@Injectable()
export class InstrumentMetadataService {
  private readonly runId = randomUUID();
  private readonly active = new Set<string>();
  private readonly producers = new Map<string, Producer>();
  private readonly queue: Producer[] = [];
  private reservations: Promise<unknown> = Promise.resolve();
  private running = false;

  constructor(
    private readonly tbank: TbankClient,
    private readonly logger: StructuredLogger,
    private readonly repository: InstrumentMetadataRepository,
    private readonly access: TbankAccessService,
  ) {}

  savedMany(identities: MetadataIdentity[]) {
    return this.repository.readMany(identities);
  }

  async read(
    userId: string,
    identity: MetadataIdentity,
    observe = false,
    revision?: string,
  ) {
    const connection = await this.access.local(userId);
    if (revision && connection.revision !== revision) return null;
    const data = (await this.savedMany([identity])).get(metadataKey(identity));
    if (!observe && !data?.fresh) {
      await this.request(userId, connection.revision, identity);
    }
    return {
      ...data,
      work: await this.repository.work(userId, connection.revision, identity),
    };
  }

  async request(
    userId: string,
    revision: string,
    identity: MetadataIdentity,
    force = false,
  ) {
    identity = {
      instrumentUid: identity.instrumentUid?.trim() || null,
      figi: identity.figi?.trim() || null,
    };
    if (!identity.instrumentUid && !identity.figi) return null;
    const normalized = identity;
    const reserved = this.reservations.then(async () => {
      const snapshot = (await this.savedMany([normalized])).get(
        metadataKey(normalized),
      );
      if (!force && snapshot?.fresh) return null;
      const result = await this.repository.reserve(
        { userId, revision },
        normalized,
        this.runId,
        this.active,
      );
      if (!result) return null;
      const { id, attemptId } = result.work;
      if (!result.launch) return id;
      const attempt = { userId, revision, id, attemptId };
      this.active.add(attemptId);
      const key = metadataKey(normalized);
      const producer = this.producers.get(key);
      if (producer) producer.observers.push(attempt);
      else {
        const queued = { identity: normalized, attempt, observers: [] };
        this.producers.set(key, queued);
        this.queue.push(queued);
      }
      if (!this.running) {
        this.running = true;
        setImmediate(() => {
          void this.drain();
        });
      }
      return id;
    });
    this.reservations = reserved.catch(() => undefined);
    return reserved;
  }

  private async drain() {
    try {
      while (this.queue.length) {
        const producer = this.queue.shift();
        if (!producer) break;
        const { attempt, identity } = producer;
        let success = false;
        try {
          const token = await this.access.tokenForAttempt(attempt);
          if (token) {
            const results: MetadataLookup[] = [];
            let useful = false;
            let uidFailed = false;
            if (identity.instrumentUid) {
              let data: null | TbankInstrument | undefined;
              try {
                data = await this.tbank.getInstrumentByUid(
                  token,
                  identity.instrumentUid,
                );
              } catch (error) {
                if (
                  !identity.figi ||
                  (error instanceof Error &&
                    isTbankErrorCode(error.message, TbankErrorCode.InvalidToken))
                ) {
                  throw error;
                }
                uidFailed = true;
              }
              if (data?.uid && data.uid !== identity.instrumentUid) {
                throw new Error('Metadata UID mismatch');
              }
              if (data !== undefined) {
                results.push({ type: 'uid', id: identity.instrumentUid, data });
              }
              useful = Boolean(data?.name?.trim() || data?.ticker?.trim());
              if (!useful && data && !identity.figi) {
                throw new Error('Incomplete metadata');
              }
            }
            if (!useful && identity.figi) {
              const data = await this.tbank.getInstrumentByFigi(
                token,
                identity.figi,
              );
              if (data?.figi && data.figi !== identity.figi) {
                throw new Error('Metadata FIGI mismatch');
              }
              if (data && !data.name?.trim() && !data.ticker?.trim()) {
                throw new Error('Incomplete metadata');
              }
              // eslint-disable-next-line sonarjs/different-types-comparison -- Defensive validation at an untrusted upstream boundary includes missing payloads.
              if (data === undefined || (uidFailed && !data)) {
                throw new Error('Metadata source unavailable');
              }
              results.push({ type: 'figi', id: identity.figi, data });
              // A successful FIGI response identifies its canonical UID too;
              // a failed UID RPC is never persisted as a negative lookup.
              if (
                uidFailed &&
                data?.uid === identity.instrumentUid &&
                identity.instrumentUid
              ) {
                results.push({ type: 'uid', id: identity.instrumentUid, data });
              }
            }
            if (
              !results.length ||
              (results.some((result) => result.data) &&
                !results.some(
                  (result) =>
                    result.data?.name?.trim() || result.data?.ticker?.trim(),
                ))
            ) {
              throw new Error('Incomplete metadata');
            }
            success = await this.repository.publish(
              attempt,
              results,
              new Date(),
            );
          }
        } catch (error) {
          if (error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)) {
            await this.access.rejectMembershipToken(attempt);
          } else {
            this.logger.event('warn', 'instrument-metadata.failed', {
              code: TbankErrorCode.Unavailable,
            });
          }
        }
        // Remove the producer before awaiting finalization: a new request cannot
        // attach to a producer whose observer list has already been drained.
        this.producers.delete(metadataKey(identity));
        for (const observer of [attempt, ...producer.observers]) {
          await this.repository.finishObserver(observer, success);
          this.active.delete(observer.attemptId);
        }
      }
    } catch {
      this.logger.event('error', 'instrument-metadata.persistence-failed', {
        code: TbankErrorCode.Unavailable,
      });
      this.active.clear();
      this.producers.clear();
      this.queue.length = 0;
    } finally {
      this.running = false;
    }
  }
}