refactor: remove unreachable catalog producer and sync orchestration (WealthLedger/backend#62) (#98)

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

@@ -3,15 +3,11 @@ import {
   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,
   tbankSourceTtlMs,
-  type TbankSyncAttempt,
 } from '../../../../integrations/tbank/tbank.api';
 import { investmentResourceMeta } from '../../investment-resource.mapper';
 import { InstrumentErrorCode } from '../domain/instrument-error-code';
@@ -31,42 +27,22 @@ import { publicInstrument } from './public-instrument';
 const UNAMBIGUOUS_MATCH_COUNT = 1;
 const NO_MATCHES = 0;
 const FIRST_RESULT_OFFSET = 0;
-const MISSING_DATA_REVISION = '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) {
+  async read(userId: string, input: CatalogQuery) {
     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({
@@ -82,7 +58,7 @@ export class CatalogDiscoveryService {
 
   async search(userId: string, input: CatalogQuery, observe = false) {
     if (!observe) return this.onlineSearch(userId, catalogQuery(input));
-    const discovery = await this.read(userId, input, observe);
+    const discovery = await this.read(userId, input);
     const sources = discovery.data
       ? await this.enrichment.readMany(
           userId,
@@ -182,11 +158,11 @@ export class CatalogDiscoveryService {
       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,
-      );
+      const alias = await this.read(userId, {
+        instrumentType,
+        query: instrument.ticker,
+        exactTicker: true,
+      });
       if (alias.data?.length === UNAMBIGUOUS_MATCH_COUNT && alias.data[0].uid === uid) {
         slug = instrument.ticker.toUpperCase();
       }
@@ -247,11 +223,11 @@ export class CatalogDiscoveryService {
       const result = await this.detail(userId, instrumentType, matches[0].uid);
       return { ...result, data: result.data ? { ...result.data, slug: query } : null };
     }
-    const lookup = await this.read(
-      userId,
-      { instrumentType, query: ticker, exactTicker: true },
-      observe,
-    );
+    const lookup = await this.read(userId, {
+      instrumentType,
+      query: ticker,
+      exactTicker: true,
+    });
     if (lookup.data?.length === NO_MATCHES) throw new NotFoundException();
     if (lookup.data && lookup.data.length > UNAMBIGUOUS_MATCH_COUNT) {
       throw new ConflictException({
@@ -337,162 +313,4 @@ export class CatalogDiscoveryService {
       }),
     };
   }
-
-  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 (
-        !force &&
-        saved.dataRevision !== MISSING_DATA_REVISION
-      ) 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;
-    }
-  }
 }

src/modules/investments/tbank-sync/tbank-sync.dto.ts

@@ -209,13 +209,6 @@ export const investmentScreenSchema = {
   },
 };
 
-export class TbankSyncAcceptedDto {
-  @ApiProperty({ format: 'uuid' }) declare id: string;
-  @ApiProperty({ type: Object, ...investmentScreenSchema })
-  declare scope: InvestmentScreenDto;
-
-  @ApiProperty({ format: 'date-time' }) declare now: string;
-}
 export class TbankSyncStatusDto {
   @ApiProperty({ format: 'uuid' }) declare id: string;
   @ApiProperty({ type: Object, ...investmentScreenSchema, nullable: true })
@@ -229,26 +222,6 @@ export class TbankSyncStatusDto {
     null | string;
 }
 
-export class HourlySyncSectionDto {
-  @ApiProperty({ enum: ['membership', 'portfolio', 'operations', 'events'] })
-  declare section: 'events' | 'membership' | 'operations' | 'portfolio';
-
-  @ApiPropertyOptional({
-    type: String,
-    nullable: false,
-    description: 'Account identifier of the problematic section, when scoped.',
-  })
-  declare accountId?: string;
-
-  @ApiProperty({
-    type: String,
-    nullable: true,
-    description:
-      'Safe error code of the failed section. No tokens or upstream payloads.',
-  })
-  declare errorCode: null | string;
-}
-
 export class HourlySyncStatusDto {
   @ApiProperty({
     enum: ['running', 'succeeded', 'partial', 'failed', 'cancelled'],

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

@@ -2,314 +2,27 @@ 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 { TbankAccessService } from '../../../integrations/tbank/tbank.api';
+import { BrokerageOperationsSyncService } from '../operations/operations.api';
+import { 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)