backend/src/modules/investments/tbank-sync/tbank-sync.service.ts

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

import { ISO_DATE_LENGTH, ISO_DATE_START } from '../../../common/time';
import { TbankErrorCode } from '../../../integrations/tbank/tbank-error-code';
import {
  InstrumentMetadataService,
  TbankAccessService,
  TbankMembershipService,
  type TbankSyncAttempt,
} from '../../../integrations/tbank/tbank.api';
import {
  CandleService,
  CatalogDiscoveryService,
  CatalogEnrichmentService,
} from '../instruments/instruments.api';
import { planInvestmentScreen } from '../investment-sync-plan';
import {
  BrokerageOperationsSyncService,
  type OperationInterval,
} from '../operations/operations.api';
import { PortfolioService } from '../portfolio/portfolio.api';
import { normalizeScreen } from './tbank-screen';
import { InvestmentScreenDto, TbankSyncStatusDto } from './tbank-sync.dto';
import {
  type ScreenIntent,
  TbankSyncRepository,
} from './tbank-sync.repository';

const FOREGROUND_CANDLE_PRIORITY = 0;

@Injectable()
export class TbankSyncService {
  private readonly runId = randomUUID();
  private readonly startedAt = new Date();
  private reservations: Promise<unknown> = Promise.resolve();
  constructor(
    private readonly repository: TbankSyncRepository,
    private readonly access: TbankAccessService,
    private readonly membership: TbankMembershipService,
    private readonly portfolio: PortfolioService,
    private readonly operations: BrokerageOperationsSyncService,
    private readonly metadata: InstrumentMetadataService,
    private readonly discovery: CatalogDiscoveryService,
    private readonly sources: CatalogEnrichmentService,
    private readonly candles: CandleService,
  ) {}

  async request(userId: string, input: InvestmentScreenDto) {
    const scope = normalizeScreen(input);
    const connection = await this.access.local(userId, scope.accountId);
    const reservation = this.reservations.then(() =>
      this.repository.reserve(userId, connection.revision, scope, this.runId),
    );
    this.reservations = reservation.catch(() => undefined);
    const reserved = await reservation;
    if (!reserved) throw new ServiceUnavailableException();
    const intent = reserved.work.intent as unknown as ScreenIntent;
    if (reserved.launch) {
      const attempt = {
        userId,
        revision: connection.revision,
        id: reserved.work.id,
        attemptId: reserved.work.attemptId,
      };
      setImmediate(() => {
        void this.dispatch(attempt, intent);
      });
    }
    return { id: reserved.work.id, ...intent };
  }

  private async dispatch(attempt: TbankSyncAttempt, intent: ScreenIntent) {
    const children: string[] = [];
    const add = async (ids: (null | string | undefined)[]) => {
      children.push(...ids.filter((id): id is string => Boolean(id)));
      await this.repository.progress(attempt, children, false);
    };
    try {
      const { userId, revision } = attempt;
      const { scope } = intent;
      const now = new Date(intent.now);
      const member = await this.membership.request(userId, revision, true);
      await add([member?.id]);
      await member?.done;
      const current = await this.access.local(userId, scope.accountId);
      if (current.revision !== revision) return;
      const accountScope = scope.accountId ?? 'all';
      const selectedAccounts = current.accounts.filter(
        (account) =>
          accountScope === 'all' || account.accountId === accountScope,
      );
      if (
        ['account', 'accounts', 'bond', 'events', 'stock'].includes(
          scope.screen,
        )
      ) {
        for (const account of scope.screen === 'stock' ||
          scope.screen === 'bond'
          ? current.accounts
          : selectedAccounts) {
          const work = await this.portfolio.request(
            userId,
            revision,
            account.accountId,
            true,
          );
          await add([work?.id]);
          await work?.done;
        }
      }
      const saved = await this.portfolio.read(
        userId,
        scope.screen === 'stock' || scope.screen === 'bond'
          ? 'all'
          : accountScope,
      );
      const positions = saved.data?.positions ?? [];
      const instruments = [
        ...new Map(
          positions.flatMap((position) =>
            position.instrumentUid &&
            (position.type === 'STOCK' || position.type === 'BOND')
              ? [
                  [
                    position.instrumentUid,
                    {
                      uid: position.instrumentUid,
                      type: position.type,
                      historyFrom: null,
                    },
                  ] as const,
                ]
              : [],
          ),
        ).values(),
      ];
      const plan = planInvestmentScreen(
        {
          ...scope,
          candleRange: scope.candleRange
            ? {
                from: new Date(scope.candleRange.from),
                to: new Date(scope.candleRange.to),
              }
            : undefined,
        },
        {
          accounts: selectedAccounts,
          instruments:
            scope.screen === 'stocks' || scope.screen === 'bonds'
              ? []
              : instruments,
          membershipKnown: Boolean(current.membershipAt),
        },
        now,
        true,
      );
      const intervals: OperationInterval[] = plan.sources.flatMap((source) =>
        source.kind === 'operations' &&
        source.accountId &&
        source.from &&
        source.to
          ? [
              {
                accountId: source.accountId,
                from: source.from,
                to: source.to,
                ...(source.instrumentUid
                  ? { instrumentUid: source.instrumentUid }
                  : {}),
              },
            ]
          : [],
      );
      if (intervals.length) {
        await add([
          await this.operations.request(
            userId,
            intervals,
            true,
            'screen',
            revision,
          ),
        ]);
      }
      for (const source of plan.sources) {
        const uid = source.instrumentUid;
        const type = source.instrumentType === 'BOND' ? 'bond' : 'share';
        if (source.kind === 'catalog') {
          await add(
            await this.discovery.refresh(userId, revision, {
              instrumentType: type,
              query: source.query ?? '',
            }),
          );
        }
        if (!uid) continue;
        if (source.kind === 'metadata') {
          await add([
            await this.metadata.request(userId, revision, {
              instrumentUid: uid,
            }),
          ]);
        }
        if (source.kind === 'quote') {
          await add(
            await this.sources.requestMany(
              userId,
              revision,
              type,
              [uid],
              true,
              [
                type === 'bond' ? 'bond-current' : 'detail',
                'last-price',
                'trading-status',
                'comparison',
              ],
            ),
          );
        }
        if (source.kind === 'company') {
          await add(
            await this.sources.requestMany(
              userId,
              revision,
              type,
              [uid],
              true,
              ['description', 'fundamentals'],
            ),
          );
        }
        if (source.kind === 'metrics') {
          await add(
            await this.sources.requestMany(
              userId,
              revision,
              type,
              [uid],
              true,
              ['coupons', 'yield', 'rating'],
            ),
          );
        }
        if (source.kind === 'calendar' && source.from && source.to) {
          await add(
            await this.sources.requestMany(
              userId,
              revision,
              type,
              [uid],
              true,
              [
                type === 'share'
                  ? 'dividends'
                  : scope.screen === 'bond'
                    ? 'coupons'
                    : 'bond-events',
              ],
              {
                from: source.from.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH),
                to: source.to.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH),
                ticker:
                  positions.find((position) => position.instrumentUid === uid)
                    ?.ticker ?? undefined,
              },
            ),
          );
        }
      }
      if (
        (scope.screen === 'stock' || scope.screen === 'bond') &&
        scope.instrumentUid
      ) {
        await add([
          await this.candles.request(
            userId,
            revision,
            {
              uid: scope.instrumentUid,
              instrumentType: scope.screen === 'bond' ? 'bond' : 'share',
              period: scope.candlePeriod ?? 'month',
              ...(scope.candleRange
                ? {
                    range: {
                      from: new Date(scope.candleRange.from),
                      to: new Date(scope.candleRange.to),
                    },
                  }
                : {}),
            },
            now,
            true,
            FOREGROUND_CANDLE_PRIORITY,
          ),
        ]);
      }
      await this.repository.progress(attempt, children, true);
    } catch {
      await this.repository
        .progress(attempt, children, true, TbankErrorCode.Unavailable)
        .catch(() => undefined);
    }
  }

  async status(userId: string, id: string): Promise<TbankSyncStatusDto> {
    const connection = await this.access
      .local(userId)
      .catch((error: unknown) => {
        if (error instanceof ConflictException) throw new NotFoundException();
        throw error;
      });
    const saved = await this.repository.read(userId, connection.revision, id);
    if (!saved) throw new NotFoundException();
    if (!saved.work.scopeKey.startsWith('screen:')) {
      return this.sourceStatus(userId, saved.work);
    }
    const intent = saved.work.intent as unknown as ScreenIntent;
    await this.access.local(userId, intent.scope.accountId);
    const children = await Promise.all(
      saved.children.map((child) =>
        child.state === 'superseded'
          ? this.operations.observe(userId, child.id)
          : Promise.resolve(child),
      ),
    );
    const running = children.some((child) => child.state === 'running');
    const failed =
      saved.work.errorCode ??
      children.find((child) => child.errorCode)?.errorCode ??
      null;
    const state =
      saved.work.processRunId !== this.runId &&
      (!saved.progress.dispatched || running)
        ? 'cancelled'
        : !saved.progress.dispatched || running
            ? 'running'
            : failed
              ? 'failed'
              : children.some((child) => child.state === 'cancelled')
                ? 'cancelled'
                : 'succeeded';
    if (
      (await this.access.local(userId, intent.scope.accountId)).revision !==
      connection.revision
    ) {
      throw new NotFoundException();
    }
    return {
      id,
      ...intent,
      state,
      errorCode: failed,
    };
  }

  private async sourceStatus(
    userId: string,
    work: NonNullable<Awaited<ReturnType<TbankSyncRepository['read']>>>['work'],
  ): Promise<TbankSyncStatusDto> {
    const observed = work.scopeKey.startsWith('operations:')
      ? await this.operations.observe(userId, work.id)
      : work;
    const state =
      observed.state === 'running'
        ? observed.updatedAt < this.startedAt
          ? 'cancelled'
          : 'running'
        : observed.state === 'failed'
          ? 'failed'
          : observed.state === 'cancelled'
            ? 'cancelled'
            : 'succeeded';
    if ((await this.access.local(userId)).revision !== work.revision) {
      throw new NotFoundException();
    }
    return {
      id: work.id,
      scope: null,
      now: work.createdAt.toISOString(),
      state,
      errorCode: observed.errorCode,
    };
  }
}