backend/src/modules/investments/instruments/application/catalog-discovery.service.ts

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

import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
import {
  TbankAccessService,
  TbankClient,
  tbankSourceNeedsRefresh,
  tbankSourceTtlMs,
  type TbankSyncAttempt,
} from '../../../../integrations/tbank/tbank.api';
import { investmentResourceMeta } from '../../investment-resource.mapper';
import { InstrumentErrorCode } from '../domain/instrument-error-code';
import {
  CatalogDiscoveryRepository,
  catalogQuery,
  type CatalogQuery,
  catalogScope,
} from '../infrastructure/catalog-discovery.repository';
import { CatalogEnrichmentService } from './catalog-enrichment.service';
import {
  catalogSearchLimit,
  selectCatalogInstruments,
} from './catalog-selection';

const UNAMBIGUOUS_MATCH_COUNT = 1;
const NO_MATCHES = 0;
const FIRST_RESULT_OFFSET = 0;

const tickerAmbiguityLimit = 2;

type Producer = {
  attempt: TbankSyncAttempt;
  force: boolean;
  input: CatalogQuery;
  observers: TbankSyncAttempt[];
};
@Injectable()
export class CatalogDiscoveryService {
  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;
  private readonly completions = new Map<
    string,
    { done: Promise<void>; resolve: () => void; workId: string }
  >();

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

  async read(userId: string, input: CatalogQuery, observe = false) {
    input = catalogQuery(input);
    const connection = await this.access.local(userId);
    const saved = await this.repository.read(input);
    if (!observe) await this.request(userId, connection.revision, input);
    return {
      data: saved.data,
      meta: investmentResourceMeta({
        resourceKey: catalogScope(input),
        dataRevision: saved.dataRevision,
        availability: saved.data === null ? 'missing' : 'complete',
        fetchedAt: saved.fetchedAt,
        freshForMs: tbankSourceTtlMs.catalog,
        work: await this.repository.work(userId, connection.revision, input),
      }),
    };
  }

  async search(userId: string, input: CatalogQuery, observe = false) {
    const discovery = await this.read(userId, input, observe);
    const sources = discovery.data
      ? await this.enrichment.readMany(
          userId,
          input.instrumentType,
          discovery.data.map((item) => item.uid),
          observe,
        )
      : [];
    const dependencies = [
      discovery.meta,
      ...sources.map((source) => source.meta),
    ];
    return {
      data:
        discovery.data?.map((short) => ({
          ...short,
          detail:
            sources.find(
              (source) =>
                source.input.uid === short.uid &&
                source.input.kind === 'detail',
            )?.instrument ?? null,
          price:
            sources.find(
              (source) =>
                source.input.uid === short.uid &&
                source.input.kind === 'last-price',
            )?.price ?? null,
        })) ?? null,
      meta: investmentResourceMeta({
        resourceKey: `search:${catalogScope(catalogQuery(input))}`,
        dataRevision: dependencies
          .map((dependency) => dependency.dataRevision)
          .join('/'),
        availability:
          discovery.data === null
            ? 'missing'
            : sources.every((source) => source.meta.availability === 'complete')
              ? 'complete'
              : 'partial',
        fetchedAt: discovery.meta.lastSuccessfulAt
          ? new Date(discovery.meta.lastSuccessfulAt)
          : null,
        freshForMs: tbankSourceTtlMs.catalog,
        work: discovery.meta.sync.id
          ? {
              id: discovery.meta.sync.id,
              state: discovery.meta.sync.state,
              errorCode: discovery.meta.sync.errorCode,
            }
          : null,
        dependencies,
      }),
    };
  }

  async detail(
    userId: string,
    instrumentType: 'bond' | 'share',
    uid: string,
    observe = false,
  ) {
    const sources = await this.enrichment.readMany(
      userId,
      instrumentType,
      [uid],
      observe,
      ['detail'],
    );
    const source = sources[0];
    if (source.meta.availability === 'complete' && !source.instrument) {
      throw new NotFoundException();
    }
    const instrument = source.instrument;
    let slug = uid;
    const dependencies = [source.meta];
    if (
      instrument?.ticker &&
      /^[A-Z0-9][\w.@+-]{0,79}$/i.test(instrument.ticker)
    ) {
      const alias = await this.read(
        userId,
        { instrumentType, query: instrument.ticker, exactTicker: true },
        observe,
      );
      if (alias.data?.length === UNAMBIGUOUS_MATCH_COUNT && alias.data[0].uid === uid) {
        slug = instrument.ticker.toUpperCase();
      }
      dependencies.push(alias.meta);
    }
    return {
      data: instrument ? { ...instrument, uid, slug } : null,
      meta: investmentResourceMeta({
        resourceKey: `${instrumentType}/detail/${uid}`,
        dataRevision: dependencies
          .map((dependency) => dependency.dataRevision)
          .join('/'),
        availability: source.meta.availability,
        fetchedAt: source.meta.lastSuccessfulAt
          ? new Date(source.meta.lastSuccessfulAt)
          : null,
        freshForMs: tbankSourceTtlMs.metadata,
        work: source.meta.sync.id
          ? {
              id: source.meta.sync.id,
              state: source.meta.sync.state,
              errorCode: source.meta.sync.errorCode,
            }
          : null,
        dependencies,
      }),
    };
  }

  async byTicker(
    userId: string,
    instrumentType: 'bond' | 'share',
    ticker: string,
    observe = false,
  ) {
    const lookup = await this.read(
      userId,
      { instrumentType, query: ticker, exactTicker: true },
      observe,
    );
    if (lookup.data?.length === NO_MATCHES) throw new NotFoundException();
    if (lookup.data && lookup.data.length > UNAMBIGUOUS_MATCH_COUNT) {
      throw new ConflictException({
        code:
          instrumentType === 'share'
            ? InstrumentErrorCode.StockTickerAmbiguous
            : InstrumentErrorCode.BondTickerAmbiguous,
      });
    }
    const result = lookup.data
      ? await this.detail(userId, instrumentType, lookup.data[0].uid, observe)
      : null;
    const dependencies = [lookup.meta, ...(result ? [result.meta] : [])];
    return {
      data: result?.data
        ? { ...result.data, slug: ticker.trim().toUpperCase() }
        : null,
      meta: investmentResourceMeta({
        resourceKey: `${instrumentType}/ticker/${ticker.trim().toUpperCase()}`,
        dataRevision: dependencies
          .map((dependency) => dependency.dataRevision)
          .join('/'),
        availability: result?.meta.availability ?? 'missing',
        fetchedAt: lookup.meta.lastSuccessfulAt
          ? new Date(lookup.meta.lastSuccessfulAt)
          : null,
        freshForMs: tbankSourceTtlMs.metadata,
        work: lookup.meta.sync.id
          ? {
              id: lookup.meta.sync.id,
              state: lookup.meta.sync.state,
              errorCode: lookup.meta.sync.errorCode,
            }
          : null,
        dependencies,
      }),
    };
  }

  request(
    userId: string,
    revision: string,
    query: CatalogQuery,
    force = false,
  ) {
    const input = catalogQuery(query);
    const reserved = this.reservations.then(async () => {
      const saved = await this.repository.read(input);
      if (
        !tbankSourceNeedsRefresh('catalog', saved.fetchedAt, new Date(), force)
      ) {
        return null;
      }
      const reservation = await this.repository.reserve(
        { userId, revision },
        input,
        this.runId,
        this.active,
      );
      if (!reservation) return null;
      const { id, attemptId } = reservation.work;
      if (!reservation.launch) {
        const current = this.producers.get(catalogScope(input));
        if (current) current.force ||= force;
        return id;
      }
      const attempt = { userId, revision, id, attemptId };
      this.active.add(attemptId);
      let resolve = () => {};
      const done = new Promise<void>((finish) => {
        resolve = finish;
      });
      this.completions.set(attemptId, { workId: id, done, resolve });
      const key = catalogScope(input);
      const producer = this.producers.get(key);
      if (producer) {
        producer.observers.push(attempt);
        producer.force ||= force;
      } else {
        const queued = { input, attempt, observers: [], force };
        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;
  }

  async refresh(userId: string, revision: string, query: CatalogQuery) {
    const id = await this.request(userId, revision, query, true);
    if (id) {
      await Promise.all(
        [...this.completions.values()]
          .filter((entry) => entry.workId === id)
          .map((entry) => entry.done),
      );
    }
    const saved = await this.repository.read(catalogQuery(query));
    const sources = await this.enrichment.readMany(
      userId,
      query.instrumentType,
      saved.data?.map((item) => item.uid) ?? [],
      true,
    );
    return [
      ...(id ? [id] : []),
      ...sources.flatMap((source) =>
        source.meta.sync.id ? [source.meta.sync.id] : [],
      ),
    ];
  }

  private async drain() {
    try {
      while (this.queue.length) {
        const producer = this.queue.shift();
        if (!producer) break;
        const { input, attempt } = producer;
        let success = false;
        try {
          const token = await this.access.tokenForAttempt(attempt);
          if (token) {
            const found =
              input.instrumentType === 'share'
                ? await this.tbank.findShares(token, input.query)
                : await this.tbank.findBonds(token, input.query);
            success = await this.repository.publish(
              attempt,
              input,
              selectCatalogInstruments(
                input.exactTicker
                  ? found.filter(
                      (item) =>
                        item.ticker?.trim().toUpperCase() === input.query,
                    )
                  : found,
                input.query,
                input.instrumentType,
              ).slice(
                FIRST_RESULT_OFFSET,
                input.exactTicker ? tickerAmbiguityLimit : catalogSearchLimit,
              ),
              new Date(),
            );
          }
        } catch (error) {
          if (error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)) {
            await this.access.rejectMembershipToken(attempt);
          } else {
            this.logger.event('warn', 'catalog-discovery.failed', {
              code: TbankErrorCode.Unavailable,
            });
          }
        }
        if (success) {
          const saved = await this.repository.read(input);
          for (const observer of [attempt, ...producer.observers]) {
            if (!input.exactTicker || saved.data?.length === UNAMBIGUOUS_MATCH_COUNT) {
              await this.enrichment.requestMany(
                observer.userId,
                observer.revision,
                input.instrumentType,
                saved.data?.map((item) => item.uid) ?? [],
                producer.force,
                input.exactTicker ? ['detail'] : ['last-price', 'detail'],
              );
            }
          }
        }
        this.producers.delete(catalogScope(input));
        for (const observer of [attempt, ...producer.observers]) {
          await this.repository.finish(observer, success);
          this.active.delete(observer.attemptId);
          this.completions.get(observer.attemptId)?.resolve();
          this.completions.delete(observer.attemptId);
        }
      }
    } catch {
      for (const completion of this.completions.values()) completion.resolve();
      this.completions.clear();
      this.active.clear();
      this.producers.clear();
      this.queue.length = 0;
      this.logger.event('error', 'catalog-discovery.persistence-failed', {
        code: TbankErrorCode.Unavailable,
      });
    } finally {
      this.running = false;
    }
  }
}