From 02e0d560c84606d6372cbac001dc94edc348995e Mon Sep 17 00:00:00 2001 From: codex-bot Date: Wed, 4 Nov 2026 17:47:36 +0300 Subject: [PATCH 1/3] refactor(api): describe bank operation create body as oneOf by kind (WealthLedger/backend#172) (#173) --- .../bank-cards/presentation/bank-card.dto.ts | 44 +++++++++++++++++++ .../presentation/bank-cards.controller.ts | 14 ++++++ 2 files changed, 58 insertions(+) diff --git a/src/modules/banking/bank-cards/presentation/bank-card.dto.ts b/src/modules/banking/bank-cards/presentation/bank-card.dto.ts index cb44a0f..defd049 100644 --- a/src/modules/banking/bank-cards/presentation/bank-card.dto.ts +++ b/src/modules/banking/bank-cards/presentation/bank-card.dto.ts @@ -1,6 +1,7 @@ import { ApiProperty, ApiPropertyOptional, + getSchemaPath, IntersectionType, OmitType, PartialType, @@ -153,6 +154,49 @@ export class CreateBankOperationDto { @MaxLength(MAX_OPERATION_DESCRIPTION_LENGTH) description?: string; } + +// OpenAPI-only variants. Runtime validation stays on CreateBankOperationDto so +// invalid combinations keep the existing BANK_INVALID_INPUT response. +class BankOperationCreateBaseDto extends OmitType(CreateBankOperationDto, [ + 'kind', 'targetCardId', 'categoryId', 'originalExpenseId', +] as const) {} + +export class CreateExpenseOperationDto extends BankOperationCreateBaseDto { + @ApiProperty({ enum: [BankOperationKind.Expense] }) declare kind: 'expense'; + @ApiPropertyOptional({ format: 'uuid' }) declare categoryId?: string; +} + +export class CreateTopupOperationDto extends BankOperationCreateBaseDto { + @ApiProperty({ enum: [BankOperationKind.Topup] }) declare kind: 'topup'; +} + +export class CreateTransferOperationDto extends BankOperationCreateBaseDto { + @ApiProperty({ enum: [BankOperationKind.Transfer] }) declare kind: 'transfer'; + @ApiProperty({ format: 'uuid' }) declare targetCardId: string; +} + +export class CreateRefundOperationDto extends BankOperationCreateBaseDto { + @ApiProperty({ enum: [BankOperationKind.Refund] }) declare kind: 'refund'; + @ApiProperty({ format: 'uuid' }) declare originalExpenseId: string; +} + +export const createBankOperationSchema = { + oneOf: [ + CreateExpenseOperationDto, + CreateTopupOperationDto, + CreateTransferOperationDto, + CreateRefundOperationDto, + ].map((model) => ({ $ref: getSchemaPath(model) })), + discriminator: { + propertyName: 'kind', + mapping: { + expense: getSchemaPath(CreateExpenseOperationDto), + topup: getSchemaPath(CreateTopupOperationDto), + transfer: getSchemaPath(CreateTransferOperationDto), + refund: getSchemaPath(CreateRefundOperationDto), + }, + }, +}; export class UpdateBankOperationDto extends PartialType( OmitType(CreateBankOperationDto, ['id'] as const), { skipNullProperties: false }, diff --git a/src/modules/banking/bank-cards/presentation/bank-cards.controller.ts b/src/modules/banking/bank-cards/presentation/bank-cards.controller.ts index 0a3e19a..2d55e2c 100644 --- a/src/modules/banking/bank-cards/presentation/bank-cards.controller.ts +++ b/src/modules/banking/bank-cards/presentation/bank-cards.controller.ts @@ -16,7 +16,9 @@ import { } from '@nestjs/common'; import { ApiBearerAuth, + ApiBody, ApiCreatedResponse, + ApiExtraModels, ApiNoContentResponse, ApiOkResponse, ApiResponse, @@ -46,6 +48,11 @@ import { CategoryNameDto, CreateBankCardDto, CreateBankOperationDto, + createBankOperationSchema, + CreateExpenseOperationDto, + CreateRefundOperationDto, + CreateTopupOperationDto, + CreateTransferOperationDto, ExpenseCategoryDto, UpdateBankCardDto, UpdateBankOperationDto, @@ -73,6 +80,12 @@ type AuthRequest = Request & { user: AccessPayload }; }, }) @ApiTags('bank-cards') +@ApiExtraModels( + CreateExpenseOperationDto, + CreateTopupOperationDto, + CreateTransferOperationDto, + CreateRefundOperationDto, +) @ApiBearerAuth() @ApiProblemResponses() @UseGuards(JwtAuthGuard) @@ -131,6 +144,7 @@ export class BankCardsController { } @Post('operations') + @ApiBody({ schema: createBankOperationSchema }) @ApiCreatedResponse({ type: BankOperationDto }) createOperation( @Req() req: AuthRequest, From c5ccfeadea7a9235ebdd6171d9b6f7b89ed2e1e7 Mon Sep 17 00:00:00 2001 From: codex-bot Date: Wed, 4 Nov 2026 18:17:09 +0300 Subject: [PATCH 2/3] refactor: extract catalog enrichment source loader (WealthLedger/backend#169) (#186) --- .../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') From 5d31b3e6dd37d1c79b61c43ed820947b5b38aaa1 Mon Sep 17 00:00:00 2001 From: codex-bot Date: Wed, 4 Nov 2026 20:47:50 +0300 Subject: [PATCH 3/3] refactor: share tbank-sync claim and generation lock (WealthLedger/backend#166) (#188) --- .../tbank/instrument-metadata.repository.ts | 35 ++++---- .../tbank/tbank-access.service.ts | 41 ++++------ src/integrations/tbank/tbank-sync-claim.ts | 53 ++++++++++++ .../infrastructure/candle.repository.ts | 82 +++++++------------ .../catalog-discovery.repository.ts | 62 +++++--------- .../brokerage-operations.repository.ts | 36 +++----- .../infrastructure/portfolio.repository.ts | 21 ++--- .../tbank-sync/tbank-sync.repository.ts | 43 ++++------ 8 files changed, 167 insertions(+), 206 deletions(-) create mode 100644 src/integrations/tbank/tbank-sync-claim.ts diff --git a/src/integrations/tbank/instrument-metadata.repository.ts b/src/integrations/tbank/instrument-metadata.repository.ts index 73a0396..8003192 100644 --- a/src/integrations/tbank/instrument-metadata.repository.ts +++ b/src/integrations/tbank/instrument-metadata.repository.ts @@ -12,6 +12,7 @@ import type { TbankInstrument } from './tbank.client'; import { PrismaService } from '../../prisma/prisma.service'; import { isJoinableRunningAttempt, type TbankSyncAttempt } from './tbank-access.service'; import { TbankErrorCode } from './tbank-error-code'; +import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim'; const FIRST_BATCH_OFFSET = 0; const UNCHANGED_REVISION_INCREMENT = 0; @@ -119,11 +120,8 @@ export class InstrumentMetadataRepository { active: ReadonlySet, ) { return this.prisma.$transaction(async (tx) => { - const valid = await tx.tbankConnection.updateMany({ - where: { ...attemptOwner, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!valid.count) return null; + const valid = await lockConnectionGeneration(tx, attemptOwner); + if (!valid) return null; const scope = { ...attemptOwner, scopeKey: `metadata:${metadataKey(identity)}`, @@ -167,15 +165,11 @@ export class InstrumentMetadataRepository { publish(attempt: TbankSyncAttempt, results: MetadataLookup[], now: Date) { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', progress: {} }, + const claimed = await claimSyncWork(tx, attempt, { + state: 'complete', + progress: {}, }); - if (!claimed.count) return false; + if (!claimed) return false; const aliases = new Map( results.map((result) => [`${result.type}:${result.id}`, result]), ); @@ -264,18 +258,17 @@ export class InstrumentMetadataRepository { } async finishObserver(attempt: TbankSyncAttempt, success: boolean) { - await this.prisma.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - ...(success ? { connection: { invalidToken: false } } : {}), - }, - data: { + await claimSyncWork( + this.prisma, + attempt, + { state: success ? 'complete' : 'failed', errorCode: success ? null : TbankErrorCode.Unavailable, }, - }); + { withoutConnectionFence: !success }, + ); if (success) { + // Fails sibling attempts whose connection was invalidated mid-flight. await this.prisma.tbankSyncWork.updateMany({ where: { ...attempt, diff --git a/src/integrations/tbank/tbank-access.service.ts b/src/integrations/tbank/tbank-access.service.ts index 98f030a..adcc5ed 100644 --- a/src/integrations/tbank/tbank-access.service.ts +++ b/src/integrations/tbank/tbank-access.service.ts @@ -12,6 +12,7 @@ import { PrismaService } from '../../prisma/prisma.service'; import { supportedTbankAccounts } from './tbank-account.policy'; import { TbankErrorCode } from './tbank-error-code'; import { tbankSourceNeedsRefresh } from './tbank-source.policy'; +import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim'; import { TbankTokenService } from './tbank-token.service'; const EXPECTED_UPDATED_CONNECTIONS = 1; @@ -56,11 +57,8 @@ export class TbankAccessService { force: boolean, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { userId, revision, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!current.count) return null; + const current = await lockConnectionGeneration(tx, { userId, revision }); + if (!current) return null; const connection = await tx.tbankConnection.findUniqueOrThrow({ where: { userId }, }); @@ -110,15 +108,11 @@ export class TbankAccessService { async failMembership(attempt: TbankSyncAttempt, invalidToken: boolean) { if (invalidToken) return this.rejectMembershipToken(attempt); - const result = await this.prisma.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'failed', errorCode: TbankErrorCode.Unavailable }, + const result = await claimSyncWork(this.prisma, attempt, { + state: 'failed', + errorCode: TbankErrorCode.Unavailable, }); - return result.count === EXPECTED_UPDATED_CONNECTIONS; + return result === EXPECTED_UPDATED_CONNECTIONS; } async tokenForAttempt(attempt: TbankSyncAttempt): Promise { @@ -163,15 +157,8 @@ export class TbankAccessService { return this.prisma.$transaction(async (tx) => { // Acquire the write fence before reading membership. Replacement and // attempt supersession cannot interleave with the publication below. - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { updatedAt: new Date() }, - }); - if (!claimed.count) return false; + const claimed = await claimSyncWork(tx, attempt, { updatedAt: new Date() }); + if (!claimed) return false; const { userId, revision } = attempt; const previous = await tx.tbankAccount.findMany({ where: { userId, revision }, @@ -244,11 +231,11 @@ export class TbankAccessService { async rejectMembershipToken(attempt: TbankSyncAttempt): Promise { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { ...attempt, state: 'running' }, - data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken }, - }); - if (!claimed.count) return false; + const claimed = await claimSyncWork(tx, attempt, { + state: 'failed', + errorCode: TbankErrorCode.InvalidToken, + }, { withoutConnectionFence: true }); + if (!claimed) return false; await tx.tbankConnection.update({ where: { userId_revision: { diff --git a/src/integrations/tbank/tbank-sync-claim.ts b/src/integrations/tbank/tbank-sync-claim.ts new file mode 100644 index 0000000..2445398 --- /dev/null +++ b/src/integrations/tbank/tbank-sync-claim.ts @@ -0,0 +1,53 @@ +import type { Prisma } from '../../../generated/prisma/client'; +import type { TbankSyncAttempt } from './tbank-access.service'; + +const NO_UPDATED_ROWS = 0; + +// Shared owners for the tbank-sync DB claim/lock protocol. +// Both helpers are CAS fences: they run inside the caller's transaction and +// keep the exact where/data shapes they replaced. Callers keep deciding +// success from the claim count exactly as before. + +// Claims the active TbankSyncWork row: only the still-running attempt of a +// valid connection may transition. `data` is the state transition to apply. +export type ClaimOptions = { + // Overrides the default 'running' fence (e.g. an initiator row also + // accepts 'complete'). + state: Prisma.TbankSyncWorkWhereInput['state'] | undefined; + // true drops the `connection: { invalidToken: false }` guard for fences + // that must also fire on an invalidated connection. + withoutConnectionFence: boolean | undefined; +}; +export async function claimSyncWork( + tx: Prisma.TransactionClient, + attempt: TbankSyncAttempt, + data: Prisma.TbankSyncWorkUpdateManyMutationInput, + options: Partial = {}, +): Promise { + const result = await tx.tbankSyncWork.updateMany({ + where: { + ...attempt, + state: options.state ?? 'running', + ...(options.withoutConnectionFence + ? {} + : { connection: { invalidToken: false } }), + }, + data, + }); + return result.count; +} + +// Locks the connection generation before reserving or publishing work: the +// row must still be the current revision with a valid token. `extra` +// (e.g. account membership) is merged into the where clause. +export async function lockConnectionGeneration( + tx: Prisma.TransactionClient, + owner: { revision: string; userId: string }, + extra: Prisma.TbankConnectionWhereInput = {}, +): Promise { + const result = await tx.tbankConnection.updateMany({ + where: { ...owner, ...extra, invalidToken: false }, + data: { membershipRevision: { increment: 0 } }, + }); + return result.count > NO_UPDATED_ROWS; +} diff --git a/src/modules/investments/instruments/infrastructure/candle.repository.ts b/src/modules/investments/instruments/infrastructure/candle.repository.ts index d01ec5b..7d308f3 100644 --- a/src/modules/investments/instruments/infrastructure/candle.repository.ts +++ b/src/modules/investments/instruments/infrastructure/candle.repository.ts @@ -2,8 +2,15 @@ import { Injectable } from '@nestjs/common'; import { createHash, randomUUID } from 'node:crypto'; import type { Prisma } from '../../../../../generated/prisma/client'; -import type { TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api'; +import { + claimSyncWork, + lockConnectionGeneration, +} from '../../../../integrations/tbank/tbank-sync-claim'; +import { + isJoinableRunningAttempt, + type TbankSyncAttempt, +} from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; import { type CandleRange, @@ -62,34 +69,30 @@ export class CandleRepository { priority = 1, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { ...owner, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!current.count) return null; + const current = await lockConnectionGeneration(tx, owner); + if (!current) return null; const scope = { ...owner, scopeKey }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: scope }, }); - if ( - previous?.state === 'running' && - previous.processRunId === runId && - active.has(previous.attemptId) - ) { - const previousIntent = previous.intent as Prisma.JsonObject; + const joinable = isJoinableRunningAttempt(previous, runId, active) + ? previous + : null; + if (joinable) { + const previousIntent = joinable.intent as Prisma.JsonObject; const previousNow = typeof previousIntent.now === 'string' ? previousIntent.now : ''; - const nextPriority = Math.min(previous.priority, priority); + const nextPriority = Math.min(joinable.priority, priority); if ( - nextPriority === previous.priority && + nextPriority === joinable.priority && (!intent.force || (previousIntent.force === true && (!intent.now || intent.now <= previousNow))) ) { - return { work: previous, launch: false }; + return { work: joinable, launch: false }; } const work = await tx.tbankSyncWork.update({ - where: { id: previous.id }, + where: { id: joinable.id }, data: { priority: nextPriority, intent: { @@ -163,15 +166,10 @@ export class CandleRepository { ).values(), ]; return this.prisma.$transaction(async (tx) => { - const valid = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { progress: { lastWindow: key } }, + const valid = await claimSyncWork(tx, attempt, { + progress: { lastWindow: key }, }); - if (!valid.count) return false; + if (!valid) return false; const identity = { workId: attempt.id, attemptId: attempt.attemptId, @@ -196,18 +194,11 @@ export class CandleRepository { completeWork = true, ) { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { - ...(completeWork ? { state: 'complete' } : {}), - errorCode: null, - }, + const claimed = await claimSyncWork(tx, attempt, { + ...(completeWork ? { state: 'complete' } : {}), + errorCode: null, }); - if (!claimed.count) return false; + if (!claimed) return false; const staged = await tx.tbankSyncStage.findMany({ where: { workId: attempt.id, @@ -311,28 +302,17 @@ export class CandleRepository { }, }); if (!origin) return false; - const result = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', errorCode: null }, + const result = await claimSyncWork(tx, attempt, { + state: 'complete', + errorCode: null, }); - return result.count > NO_UPDATED_ROWS; + return result > NO_UPDATED_ROWS; }); } fail(attempt: TbankSyncAttempt, errorCode: string) { return this.prisma.$transaction(async (tx) => { - await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'failed', errorCode }, - }); + await claimSyncWork(tx, attempt, { state: 'failed', errorCode }); await tx.tbankSyncStage.deleteMany({ where: { workId: attempt.id, attemptId: attempt.attemptId }, }); diff --git a/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts b/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts index e560e20..8f508c3 100644 --- a/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts +++ b/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts @@ -5,6 +5,7 @@ import { isDeepStrictEqual } from 'node:util'; import type { Prisma } from '../../../../../generated/prisma/client'; import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; +import { claimSyncWork, lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim'; import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; @@ -159,24 +160,18 @@ export class CatalogDiscoveryRepository { initiator: TbankSyncAttempt = attempt, ) { return this.prisma.$transaction(async (tx) => { - const origin = await tx.tbankSyncWork.updateMany({ - where: { - ...initiator, - state: { in: ['running', 'complete'] }, - connection: { invalidToken: false }, - }, - data: { progress: {} }, + const origin = await claimSyncWork( + tx, + initiator, + { progress: {} }, + { state: { in: ['running', 'complete'] } }, + ); + if (!origin) return false; + const claimed = await claimSyncWork(tx, attempt, { + state: 'complete', + progress: {}, }); - if (!origin.count) return false; - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', progress: {} }, - }); - if (!claimed.count) return false; + if (!claimed) return false; const payload = JSON.parse( JSON.stringify(value), ) as Prisma.InputJsonObject; @@ -284,11 +279,8 @@ export class CatalogDiscoveryRepository { active: ReadonlySet, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { ...owner, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!current.count) return null; + const current = await lockConnectionGeneration(tx, owner); + if (!current) return null; const scope = { ...owner, scopeKey: catalogScope(input) }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: scope }, @@ -341,15 +333,11 @@ export class CatalogDiscoveryRepository { now: Date, ) { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', progress: {} }, + const claimed = await claimSyncWork(tx, attempt, { + state: 'complete', + progress: {}, }); - if (!claimed.count) return false; + if (!claimed) return false; for (const instrument of instruments) { // Discovery can create a minimal identity, but cannot overwrite verified metadata. const identity = { provider, uid: instrument.uid }; @@ -461,6 +449,7 @@ export class CatalogDiscoveryRepository { finish(attempt: TbankSyncAttempt, success: boolean) { return this.prisma.$transaction(async (tx) => { + // Fails sibling attempts whose connection was invalidated mid-flight. await tx.tbankSyncWork.updateMany({ where: { ...attempt, @@ -469,16 +458,9 @@ export class CatalogDiscoveryRepository { }, data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken }, }); - await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { - state: success ? 'complete' : 'failed', - errorCode: success ? null : TbankErrorCode.Unavailable, - }, + await claimSyncWork(tx, attempt, { + state: success ? 'complete' : 'failed', + errorCode: success ? null : TbankErrorCode.Unavailable, }); }); } diff --git a/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts b/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts index 844d1a8..161661f 100644 --- a/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts +++ b/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts @@ -7,13 +7,14 @@ import { createHash, randomUUID } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import type { Prisma } from '../../../../../generated/prisma/client'; -import type { - TbankOperationItem, - TbankOperationsByCursorResponse, -} from '../../../../integrations/tbank/tbank.api'; import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; import { isTbankProtocolValue, TbankOperationState } from '../../../../integrations/tbank/tbank-protocol'; +import { claimSyncWork, lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim'; +import { + isJoinableRunningAttempt, + type TbankOperationItem, type TbankOperationsByCursorResponse, +} from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; import { dateOnlyBounds, @@ -268,16 +269,10 @@ export class BrokerageOperationsRepository { .update(JSON.stringify([...new Set(intervals.map(intervalKey))].sort())) .digest('hex'); return this.prisma.$transaction(async (tx) => { - const locked = await tx.tbankConnection.updateMany({ - where: { - userId, - revision, - invalidToken: false, - membershipAt: { not: null }, - }, - data: { membershipRevision: { increment: 0 } }, + const locked = await lockConnectionGeneration(tx, { userId, revision }, { + membershipAt: { not: null }, }); - if (!locked.count) return null; + if (!locked) return null; const accounts = await tx.tbankAccount.findMany({ where: { userId, revision }, select: { accountId: true, openedOn: true }, @@ -379,9 +374,7 @@ export class BrokerageOperationsRepository { publicationAtomic && previousIntent && !previousIntent.atomic; if ( touching.length <= SINGLE_WORK_GROUP && - previous?.state === 'running' && - previous.processRunId === processRunId && - runningAttempts.has(previous.attemptId) && + isJoinableRunningAttempt(previous, processRunId, runningAttempts) && !upgradeGroup ) { const pending = previousIntent?.intervals.map(decodeInterval) ?? []; @@ -933,15 +926,8 @@ async function claim( tx: Prisma.TransactionClient, attempt: OperationAttempt, ): Promise { - const result = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { updatedAt: new Date() }, - }); - return result.count === EXPECTED_UPDATED_WORK_ITEMS; + const count = await claimSyncWork(tx, attempt, { updatedAt: new Date() }); + return count === EXPECTED_UPDATED_WORK_ITEMS; } function decodeInterval(key: string): OperationInterval { diff --git a/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts b/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts index 3be02a0..1955ba5 100644 --- a/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts +++ b/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts @@ -5,7 +5,8 @@ import type { Prisma } from '../../../../../generated/prisma/client'; import type { PortfolioDto } from '../presentation/portfolio.dto'; import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; -import { tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api'; +import { lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim'; +import { isJoinableRunningAttempt, tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; const EXPECTED_UPDATED_ACCOUNTS = 1; @@ -56,16 +57,10 @@ export class PortfolioRepository { force: boolean, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { - userId, - revision, - invalidToken: false, - accounts: { some: { accountId } }, - }, - data: { membershipRevision: { increment: 0 } }, + const current = await lockConnectionGeneration(tx, { userId, revision }, { + accounts: { some: { accountId } }, }); - if (!current.count) return null; + if (!current) return null; const snapshot = await tx.tbankPortfolio.findUnique({ where: { userId_revision_accountId: { userId, revision, accountId } }, }); @@ -86,11 +81,7 @@ export class PortfolioRepository { return null; } const scope = portfolioWorkScope; - if ( - previous?.state === 'running' && - previous.processRunId === processRunId && - pending.has(previous.attemptId) - ) { + if (isJoinableRunningAttempt(previous, processRunId, pending)) { return { work: previous, launch: false }; } const data = { diff --git a/src/modules/investments/tbank-sync/tbank-sync.repository.ts b/src/modules/investments/tbank-sync/tbank-sync.repository.ts index 8ccb815..fb1a872 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.repository.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.repository.ts @@ -6,6 +6,7 @@ import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api'; import type { InvestmentAttemptDto } from '../investment-resource.dto'; import type { InvestmentScreenDto } from './tbank-sync.dto'; +import { claimSyncWork, lockConnectionGeneration } from '../../../integrations/tbank/tbank-sync-claim'; import { PrismaService } from '../../../prisma/prisma.service'; export type ScreenIntent = { now: string; scope: InvestmentScreenDto }; @@ -88,11 +89,8 @@ export class TbankSyncRepository { runId: string, ) { return this.prisma.$transaction(async (tx) => { - const valid = await tx.tbankConnection.updateMany({ - where: { userId, revision, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!valid.count) return null; + const valid = await lockConnectionGeneration(tx, { userId, revision }); + if (!valid) return null; const scopeKey = `screen:${JSON.stringify(scope)}`; const key = { userId, revision, scopeKey }; const previous = await tx.tbankSyncWork.findUnique({ @@ -176,17 +174,11 @@ export class TbankSyncRepository { ...Object.fromEntries(rows.map((row) => [row.id, row.attemptId])), ...old.attempts, }; - return tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { - progress: { children: [...new Set(children)], attempts, dispatched }, - ...(errorCode ? { state: 'failed', errorCode } : {}), - }, + const claimed = await claimSyncWork(tx, attempt, { + progress: { children: [...new Set(children)], attempts, dispatched }, + ...(errorCode ? { state: 'failed', errorCode } : {}), }); + return { count: claimed }; }); } @@ -364,20 +356,17 @@ export class TbankSyncRepository { where: { userId: attempt.userId, id: child.id }, select: { attemptId: true }, }); - const updated = await tx.tbankSyncWork.updateMany({ - where: { ...attempt, state: 'running' }, - data: { - progress: { - attempts: { - ...progress.attempts, - ...(row ? { [child.id]: row.attemptId } : {}), - }, - children: [...(progress.children ?? []), child], - sections: progress.sections ?? [], + const updated = await claimSyncWork(tx, attempt, { + progress: { + attempts: { + ...progress.attempts, + ...(row ? { [child.id]: row.attemptId } : {}), }, + children: [...(progress.children ?? []), child], + sections: progress.sections ?? [], }, - }); - return Boolean(updated.count); + }, { withoutConnectionFence: true }); + return Boolean(updated); }); }