Merge branch 'develop' into refactor/167-split-brokerage-operations

2026-11-04 21:14:38 +03:00
parent 5aa52586fd 5d31b3e6dd
commit 18c5afc6f1
14 changed files with 460 additions and 417 deletions
@@ -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<string>,
  ) {
    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,
@@ -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<null | string> {
@@ -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<boolean> {
    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: {
@@ -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<ClaimOptions> = {},
+): Promise<number> {
+  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<boolean> {
+  const result = await tx.tbankConnection.updateMany({
+    where: { ...owner, ...extra, invalidToken: false },
+    data: { membershipRevision: { increment: 0 } },
+  });
+  return result.count > NO_UPDATED_ROWS;
+}
@@ -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 },
@@ -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,
@@ -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<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;
@@ -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');
+  }
+}
@@ -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 },
      });
@@ -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<string>,
  ) {
    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,
      });
    });
  }
@@ -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,
  ],
@@ -3,13 +3,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 {
  decodeInterval,
@@ -88,16 +89,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 },
@@ -199,9 +194,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) ?? [];
@@ -706,13 +699,6 @@ async function claim(
  tx: Prisma.TransactionClient,
  attempt: OperationAttempt,
 ): Promise<boolean> {
-  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;
 }
@@ -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 = {
@@ -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);
    });
  }

@@ -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')