refactor: extract catalog enrichment source loader (WealthLedger/backend#169)

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

@@ -1,43 +1,25 @@
 import { Injectable, NotFoundException } from '@nestjs/common';
 import { randomUUID } from 'node:crypto';
 
-import {
-  decimalFromString,
-  divide,
-} from '../../../../common/decimal';
 import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
-import {
-  MILLISECONDS_PER_DAY,
-  MILLISECONDS_PER_SECOND,
-} from '../../../../common/time';
-import { type MoexBondEvent, MoexClient } from '../../../../integrations/moex/moex.api';
-import { NcrRatingsClient } from '../../../../integrations/ncr/ncr.api';
 import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
 import {
   TbankAccessService,
-  type TbankBond,
-  type TbankCandle,
   TbankClient,
-  type TbankCoupon,
+  TbankLastPrice,
-  type TbankDividend,
-  type TbankFundamentals,
-  type TbankLastPrice,
-  type TbankShare,
   tbankSourceNeedsRefresh,
   tbankSourceTtlMs,
   type TbankSyncAttempt,
 } from '../../../../integrations/tbank/tbank.api';
 import { investmentResourceMeta } from '../../investment-resource.mapper';
-import { candleGaps, utcDay } from '../domain/candle-coverage';
-import { CandleRepository } from '../infrastructure/candle.repository';
 import {
   CatalogDiscoveryRepository,
   catalogScope,
   type CatalogSource,
 } from '../infrastructure/catalog-discovery.repository';
-import { toSavedCandle } from '../infrastructure/tbank-candle.mapper';
+import { type CatalogPayload, CatalogSourceLoader } from './catalog-source.loader';
+export type { CatalogPayload } from './catalog-source.loader';
 import { catalogSearchLimit } from './catalog-selection';
-import { previousClose, timestamp } from './instrument-market';
 import { publicInstrument } from './public-instrument';
 
 const MISSING_DATA_REVISION = 0;
@@ -93,9 +75,7 @@ export class CatalogEnrichmentService {
     private readonly access: TbankAccessService,
     private readonly tbank: TbankClient,
     private readonly logger: StructuredLogger,
-    private readonly moex: MoexClient,
+    private readonly loader: CatalogSourceLoader,
-    private readonly ratings: NcrRatingsClient,
-    private readonly candles: CandleRepository,
   ) {}
 
   async readMany(
@@ -337,169 +317,6 @@ export class CatalogEnrichmentService {
     };
   }
 
-  // eslint-disable-next-line sonarjs/cognitive-complexity -- Source loading intentionally dispatches by persisted resource kind.
-  private async load(
-    input: CatalogSource,
-    token: string,
-    attempt: TbankSyncAttempt,
-  ): Promise<CatalogPayload> {
-    if (input.kind === 'detail' || input.kind === 'bond-current') {
-      const instrument =
-        input.instrumentType === 'share'
-          ? await this.tbank.getShare(token, input.uid)
-          : await this.tbank.getBond(token, input.uid);
-      if (instrument && instrument.uid !== input.uid) {
-        throw new Error('Catalog UID mismatch');
-      }
-      // eslint-disable-next-line sonarjs/different-types-comparison -- Defensive validation rejects missing upstream payloads instead of caching them as absence.
-      if (instrument === undefined) throw new Error('Missing catalog response');
-      return { instrument: instrument ? publicInstrument({ ...instrument, uid: input.uid }) : null };
-    }
-    if (input.kind === 'trading-status') {
-      return {
-        tradingStatus: await this.tbank.getTradingStatus(token, input.uid),
-      };
-    }
-    if (input.kind === 'comparison') {
-      const prices = await this.repository.readSources([
-        { ...input, kind: 'last-price', rangeKey: undefined },
-      ]);
-      if (!prices.length) throw new Error('Price dependency missing');
-      const price = (prices[0].payload as CatalogPayload).price;
-      const priceAt = timestamp(price?.time);
-      if (!priceAt) return { candle: null };
-      const now = new Date();
-      const range = {
-        from: utcDay(new Date(Date.parse(priceAt) - previousCloseLookbackMs)),
-        to: new Date(priceAt),
-      };
-      const saved = await this.candles.read(input.uid, range);
-      const gaps = candleGaps(range, saved.coverage, now);
-      for (const gap of gaps) {
-        const response = await this.tbank.getCandles(
-          token,
-          input.uid,
-          gap.from,
-          gap.to,
-        );
-        const rows = response.map(toSavedCandle);
-        if (!(await this.candles.stage(attempt, gap, rows, now))) {
-          throw new Error('Cancelled candle comparison');
-        }
-      }
-      if (
-        gaps.length &&
-        !(await this.candles.publish(
-          attempt,
-          { uid: input.uid, ranges: gaps },
-          false,
-        ))
-      ) {
-        throw new Error('Cancelled candle comparison');
-      }
-      const values = (await this.candles.read(input.uid, range)).data.map(
-        (row) => {
-          const value = divide(
-            decimalFromString(row.close),
-            decimalFromString('1'),
-            quotationScale,
-          ).value;
-          return {
-            time: {
-              seconds: String(Date.parse(row.time) / MILLISECONDS_PER_SECOND),
-            },
-            close: {
-              units: String(value / quotationUnits),
-              nano: Number(value % quotationUnits),
-            },
-            isComplete: row.isComplete,
-          };
-        },
-      );
-      return { candle: previousClose(values, priceAt) };
-    }
-    const details = await this.repository.readSources([
-      { ...input, kind: 'detail', rangeKey: undefined },
-      ...(input.instrumentType === 'bond'
-        ? [{ ...input, kind: 'bond-current' as const, rangeKey: undefined }]
-        : []),
-    ]);
-    const instrument = details
-      .map((row) => (row.payload as CatalogPayload).instrument)
-      .find(Boolean);
-    if (
-      !instrument &&
-      !['bond-events', 'coupons', 'dividends'].includes(input.kind)
-    ) {
-      throw new Error('Instrument dependency missing');
-    }
-    if (input.kind === 'description') {
-      return {
-        description: instrument?.assetUid
-          ? await this.tbank.getAssetDescription(token, instrument?.assetUid)
-          : null,
-      };
-    }
-    if (input.kind === 'fundamentals') {
-      return {
-        fundamentals: instrument?.assetUid
-          ? await this.tbank.getFundamentals(token, instrument?.assetUid)
-          : null,
-      };
-    }
-    if (input.kind === 'coupons' || input.kind === 'dividends') {
-      const year = Number(input.rangeKey);
-      const [from, to] = input.rangeKey?.startsWith('[')
-        ? (JSON.parse(input.rangeKey) as [string, string])
-        : [
-            `${year - payoutHistoryYears}-01-01`,
-            `${year + payoutFutureYears}-01-01`,
-          ];
-      return input.kind === 'coupons'
-        ? {
-            coupons: await this.tbank.getBondCoupons(
-              token,
-              input.uid,
-              from,
-              to,
-            ),
-          }
-        : {
-            dividends: await this.tbank.getDividends(
-              token,
-              input.uid,
-              from,
-              to,
-            ),
-          };
-    }
-    if (input.kind === 'bond-events') {
-      const ticker = instrument?.ticker || input.ticker;
-      if (!ticker) throw new Error('Bond ticker missing');
-      return { bondEvents: await this.moex.bondEvents(ticker) };
-    }
-    if (input.kind === 'yield') {
-      return {
-        effectiveYield:
-          instrument?.ticker && instrument.isin && instrument.classCode
-            ? await this.moex.bondYield(
-                instrument.ticker,
-                instrument.isin,
-                instrument.classCode,
-              )
-            : null,
-      };
-    }
-    if (input.kind === 'rating') {
-      return {
-        creditRating: instrument?.isin
-          ? await this.ratings.rating(instrument.isin)
-          : null,
-      };
-    }
-    throw new Error('Unsupported catalog source');
-  }
-
   // eslint-disable-next-line sonarjs/cognitive-complexity -- Queue draining coordinates publication fences and observer completion.
   private async drain() {
     try {
@@ -537,7 +354,7 @@ export class CatalogEnrichmentService {
               batch.map((item) => item.input.uid),
             );
           } else {
-            payload = await this.load(producer.input, token, producer.attempt);
+            payload = await this.loader.load(producer.input, token, producer.attempt);
           }
         } catch (error) {
           failed = true;
@@ -639,24 +456,6 @@ function sourceInputs(
     })),
   );
 }
-const previousCloseLookbackDays = 20;
-const previousCloseLookbackMs =
-  previousCloseLookbackDays * MILLISECONDS_PER_DAY;
-const payoutHistoryYears = 5;
-const payoutFutureYears = 2;
-export type CatalogPayload = {
-  bondEvents?: MoexBondEvent[];
-  candle?: null | TbankCandle;
-  coupons?: TbankCoupon[];
-  creditRating?: Awaited<ReturnType<NcrRatingsClient['rating']>>;
-  description?: null | string;
-  dividends?: TbankDividend[];
-  effectiveYield?: Awaited<ReturnType<MoexClient['bondYield']>>;
-  fundamentals?: null | TbankFundamentals;
-  instrument?: null | TbankBond | TbankShare;
-  price?: null | TbankLastPrice;
-  tradingStatus?: null | string;
-};
 
 function sourcePolicy(kind: CatalogSource['kind']) {
   if (
@@ -679,6 +478,3 @@ function producerKey(input: CatalogSource) {
     ? `bond-metadata:${input.uid}`
     : catalogScope(input);
 }
-
-const quotationScale = 9;
-const quotationUnits = 1_000_000_000n;

src/modules/investments/instruments/application/catalog-source.loader.ts

@@ -0,0 +1,228 @@
+import { Injectable } from '@nestjs/common';
+
+import {
+  decimalFromString,
+  divide,
+} from '../../../../common/decimal';
+import {
+  MILLISECONDS_PER_DAY,
+  MILLISECONDS_PER_SECOND,
+} from '../../../../common/time';
+import { type MoexBondEvent, MoexClient } from '../../../../integrations/moex/moex.api';
+import { NcrRatingsClient } from '../../../../integrations/ncr/ncr.api';
+import {
+  type TbankBond,
+  type TbankCandle,
+  TbankClient,
+  type TbankCoupon,
+  type TbankDividend,
+  type TbankFundamentals,
+  type TbankLastPrice,
+  type TbankShare,
+  type TbankSyncAttempt,
+} from '../../../../integrations/tbank/tbank.api';
+import { candleGaps, utcDay } from '../domain/candle-coverage';
+import { CandleRepository } from '../infrastructure/candle.repository';
+import {
+  CatalogDiscoveryRepository,
+  type CatalogSource,
+} from '../infrastructure/catalog-discovery.repository';
+import { toSavedCandle } from '../infrastructure/tbank-candle.mapper';
+import { previousClose, timestamp } from './instrument-market';
+import { publicInstrument } from './public-instrument';
+
+export type CatalogPayload = {
+  bondEvents?: MoexBondEvent[];
+  candle?: null | TbankCandle;
+  coupons?: TbankCoupon[];
+  creditRating?: Awaited<ReturnType<NcrRatingsClient['rating']>>;
+  description?: null | string;
+  dividends?: TbankDividend[];
+  effectiveYield?: Awaited<ReturnType<MoexClient['bondYield']>>;
+  fundamentals?: null | TbankFundamentals;
+  instrument?: null | TbankBond | TbankShare;
+  price?: null | TbankLastPrice;
+  tradingStatus?: null | string;
+};
+
+const previousCloseLookbackDays = 20;
+const previousCloseLookbackMs =
+  previousCloseLookbackDays * MILLISECONDS_PER_DAY;
+const payoutHistoryYears = 5;
+const payoutFutureYears = 2;
+const quotationScale = 9;
+const quotationUnits = 1_000_000_000n;
+
+@Injectable()
+export class CatalogSourceLoader {
+  constructor(
+    private readonly repository: CatalogDiscoveryRepository,
+    private readonly tbank: TbankClient,
+    private readonly moex: MoexClient,
+    private readonly ratings: NcrRatingsClient,
+    private readonly candles: CandleRepository,
+  ) {}
+
+  // eslint-disable-next-line sonarjs/cognitive-complexity -- Source loading intentionally dispatches by persisted resource kind.
+  async load(
+    input: CatalogSource,
+    token: string,
+    attempt: TbankSyncAttempt,
+  ): Promise<CatalogPayload> {
+    if (input.kind === 'detail' || input.kind === 'bond-current') {
+      const instrument =
+        input.instrumentType === 'share'
+          ? await this.tbank.getShare(token, input.uid)
+          : await this.tbank.getBond(token, input.uid);
+      if (instrument && instrument.uid !== input.uid) {
+        throw new Error('Catalog UID mismatch');
+      }
+      // eslint-disable-next-line sonarjs/different-types-comparison -- Defensive validation rejects missing upstream payloads instead of caching them as absence.
+      if (instrument === undefined) throw new Error('Missing catalog response');
+      return { instrument: instrument ? publicInstrument({ ...instrument, uid: input.uid }) : null };
+    }
+    if (input.kind === 'trading-status') {
+      return {
+        tradingStatus: await this.tbank.getTradingStatus(token, input.uid),
+      };
+    }
+    if (input.kind === 'comparison') {
+      const prices = await this.repository.readSources([
+        { ...input, kind: 'last-price', rangeKey: undefined },
+      ]);
+      if (!prices.length) throw new Error('Price dependency missing');
+      const price = (prices[0].payload as CatalogPayload).price;
+      const priceAt = timestamp(price?.time);
+      if (!priceAt) return { candle: null };
+      const now = new Date();
+      const range = {
+        from: utcDay(new Date(Date.parse(priceAt) - previousCloseLookbackMs)),
+        to: new Date(priceAt),
+      };
+      const saved = await this.candles.read(input.uid, range);
+      const gaps = candleGaps(range, saved.coverage, now);
+      for (const gap of gaps) {
+        const response = await this.tbank.getCandles(
+          token,
+          input.uid,
+          gap.from,
+          gap.to,
+        );
+        const rows = response.map(toSavedCandle);
+        if (!(await this.candles.stage(attempt, gap, rows, now))) {
+          throw new Error('Cancelled candle comparison');
+        }
+      }
+      if (
+        gaps.length &&
+        !(await this.candles.publish(
+          attempt,
+          { uid: input.uid, ranges: gaps },
+          false,
+        ))
+      ) {
+        throw new Error('Cancelled candle comparison');
+      }
+      const values = (await this.candles.read(input.uid, range)).data.map(
+        (row) => {
+          const value = divide(
+            decimalFromString(row.close),
+            decimalFromString('1'),
+            quotationScale,
+          ).value;
+          return {
+            time: {
+              seconds: String(Date.parse(row.time) / MILLISECONDS_PER_SECOND),
+            },
+            close: {
+              units: String(value / quotationUnits),
+              nano: Number(value % quotationUnits),
+            },
+            isComplete: row.isComplete,
+          };
+        },
+      );
+      return { candle: previousClose(values, priceAt) };
+    }
+    const details = await this.repository.readSources([
+      { ...input, kind: 'detail', rangeKey: undefined },
+      ...(input.instrumentType === 'bond'
+        ? [{ ...input, kind: 'bond-current' as const, rangeKey: undefined }]
+        : []),
+    ]);
+    const instrument = details
+      .map((row) => (row.payload as CatalogPayload).instrument)
+      .find(Boolean);
+    if (
+      !instrument &&
+      !['bond-events', 'coupons', 'dividends'].includes(input.kind)
+    ) {
+      throw new Error('Instrument dependency missing');
+    }
+    if (input.kind === 'description') {
+      return {
+        description: instrument?.assetUid
+          ? await this.tbank.getAssetDescription(token, instrument?.assetUid)
+          : null,
+      };
+    }
+    if (input.kind === 'fundamentals') {
+      return {
+        fundamentals: instrument?.assetUid
+          ? await this.tbank.getFundamentals(token, instrument?.assetUid)
+          : null,
+      };
+    }
+    if (input.kind === 'coupons' || input.kind === 'dividends') {
+      const year = Number(input.rangeKey);
+      const [from, to] = input.rangeKey?.startsWith('[')
+        ? (JSON.parse(input.rangeKey) as [string, string])
+        : [
+            `${year - payoutHistoryYears}-01-01`,
+            `${year + payoutFutureYears}-01-01`,
+          ];
+      return input.kind === 'coupons'
+        ? {
+            coupons: await this.tbank.getBondCoupons(
+              token,
+              input.uid,
+              from,
+              to,
+            ),
+          }
+        : {
+            dividends: await this.tbank.getDividends(
+              token,
+              input.uid,
+              from,
+              to,
+            ),
+          };
+    }
+    if (input.kind === 'bond-events') {
+      const ticker = instrument?.ticker || input.ticker;
+      if (!ticker) throw new Error('Bond ticker missing');
+      return { bondEvents: await this.moex.bondEvents(ticker) };
+    }
+    if (input.kind === 'yield') {
+      return {
+        effectiveYield:
+          instrument?.ticker && instrument.isin && instrument.classCode
+            ? await this.moex.bondYield(
+                instrument.ticker,
+                instrument.isin,
+                instrument.classCode,
+              )
+            : null,
+      };
+    }
+    if (input.kind === 'rating') {
+      return {
+        creditRating: instrument?.isin
+          ? await this.ratings.rating(instrument.isin)
+          : null,
+      };
+    }
+    throw new Error('Unsupported catalog source');
+  }
+}

src/modules/investments/instruments/instrument-sources.module.ts

@@ -6,6 +6,7 @@ import { TbankModule } from '../../../integrations/tbank/tbank.module';
 import { CandleService } from './application/candle.service';
 import { CatalogDiscoveryService } from './application/catalog-discovery.service';
 import { CatalogEnrichmentService } from './application/catalog-enrichment.service';
+import { CatalogSourceLoader } from './application/catalog-source.loader';
 import { CandleRepository } from './infrastructure/candle.repository';
 import { CatalogDiscoveryRepository } from './infrastructure/catalog-discovery.repository';
 
@@ -15,6 +16,7 @@ import { CatalogDiscoveryRepository } from './infrastructure/catalog-discovery.r
     CandleService,
     CandleRepository,
     CatalogEnrichmentService,
+    CatalogSourceLoader,
     CatalogDiscoveryService,
     CatalogDiscoveryRepository,
   ],

test/tbank-hourly-lifecycle.e2e-spec.ts

@@ -640,8 +640,6 @@ describe('hourly attempt lifecycle (SQLite)', () => {
       { getLastPrices: jest.fn().mockResolvedValue([]) } as never,
       logger as never,
       {} as never,
-      {} as never,
-      {} as never,
     );
     jest
       .spyOn(catalogRepository, 'publishSource')