From 95ba1578211f0a5e8c7223f9a962fb5bf3bcce52 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Wed, 4 Nov 2026 01:29:30 +0300 Subject: [PATCH] refactor: extract catalog enrichment source loader (WealthLedger/backend#169) --- .../application/catalog-enrichment.service.ts | 214 +--------------- .../application/catalog-source.loader.ts | 228 ++++++++++++++++++ .../instruments/instrument-sources.module.ts | 2 + test/tbank-hourly-lifecycle.e2e-spec.ts | 2 - 4 files changed, 235 insertions(+), 211 deletions(-) create mode 100644 src/modules/investments/instruments/application/catalog-source.loader.ts diff --git a/src/modules/investments/instruments/application/catalog-enrichment.service.ts b/src/modules/investments/instruments/application/catalog-enrichment.service.ts index 217a77b..a4da009 100644 --- a/src/modules/investments/instruments/application/catalog-enrichment.service.ts +++ b/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, - type TbankDividend, - type TbankFundamentals, - type TbankLastPrice, - type TbankShare, + TbankLastPrice, 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 ratings: NcrRatingsClient, - private readonly candles: CandleRepository, + private readonly loader: CatalogSourceLoader, ) {} 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 { - 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>; - description?: null | string; - dividends?: TbankDividend[]; - effectiveYield?: Awaited>; - 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; diff --git a/src/modules/investments/instruments/application/catalog-source.loader.ts b/src/modules/investments/instruments/application/catalog-source.loader.ts new file mode 100644 index 0000000..e3b9f08 --- /dev/null +++ b/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>; + description?: null | string; + dividends?: TbankDividend[]; + effectiveYield?: Awaited>; + 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 { + 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'); + } +} diff --git a/src/modules/investments/instruments/instrument-sources.module.ts b/src/modules/investments/instruments/instrument-sources.module.ts index 23e20cf..6479c41 100644 --- a/src/modules/investments/instruments/instrument-sources.module.ts +++ b/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, ], diff --git a/test/tbank-hourly-lifecycle.e2e-spec.ts b/test/tbank-hourly-lifecycle.e2e-spec.ts index b9bdae5..60b0988 100644 --- a/test/tbank-hourly-lifecycle.e2e-spec.ts +++ b/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')