refactor: tidy tbank-sync claim comments (WealthLedger/backend#166)

.gitea/workflows/verify.yml

@@ -67,7 +67,7 @@ jobs:
       - name: Audit runtime and development dependencies
         run: |
           npm audit --omit=dev --audit-level=low
-          npm audit --include=dev --audit-level=high --ignore-advisory=GHSA-hp3w-g68c-fv3c
+          npm audit --include=dev --audit-level=high
 
   verify:
     needs: [commits, security]

renovate.json

@@ -2,7 +2,8 @@
   "$schema": "https://docs.renovatebot.com/renovate-schema.json",
   "extends": [
     "config:recommended",
-    ":semanticCommitTypeAll(chore)"
+    ":semanticCommitTypeAll(chore)",
+    "security:minimumReleaseAgeNpm"
   ],
   "timezone": "Europe/Moscow",
   "labels": [
@@ -10,6 +11,13 @@
   ],
   "dependencyDashboard": true,
   "rangeStrategy": "pin",
+  "prConcurrentLimit": 5,
+  "lockFileMaintenance": {
+    "enabled": true,
+    "schedule": [
+      "before 4am on the first day of the month"
+    ]
+  },
   "packageRules": [
     {
       "matchUpdateTypes": [
@@ -18,6 +26,27 @@
       ],
       "groupName": "non-major dependencies"
     },
+    {
+      "matchManagers": [
+        "github-actions"
+      ],
+      "matchUpdateTypes": [
+        "minor",
+        "patch"
+      ],
+      "groupName": "non-major github actions"
+    },
+    {
+      "matchManagers": [
+        "docker-compose",
+        "dockerfile"
+      ],
+      "matchUpdateTypes": [
+        "minor",
+        "patch"
+      ],
+      "groupName": "non-major container images"
+    },
     {
       "matchUpdateTypes": [
         "major"

src/integrations/tbank/instrument-metadata.repository.ts

@@ -7,11 +7,12 @@ import type {
   TbankInstrument as MetadataRow,
   Prisma,
 } from '../../../generated/prisma/client';
-import type { TbankSyncAttempt } from './tbank-access.service';
 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)}`,
@@ -131,11 +129,7 @@ export class InstrumentMetadataRepository {
       const previous = await tx.tbankSyncWork.findUnique({
         where: { userId_revision_scopeKey: scope },
       });
-      if (
-        previous?.state === 'running' &&
-        previous.processRunId === runId &&
-        active.has(previous.attemptId)
-      ) {
+      if (isJoinableRunningAttempt(previous, runId, active)) {
         return { work: previous, launch: false };
       }
       const data = {
@@ -171,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]),
       );
@@ -268,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,

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;
@@ -25,6 +26,22 @@ export type TbankSyncAttempt = {
   userId: string;
 };
 
+// Join rule shared by sync reservations: the same process joins its own running attempt
+// only while that attempt is still active in memory. Set and Map both satisfy `has`.
+export function isJoinableRunningAttempt<
+  T extends { attemptId: string; processRunId: string; state: string },
+>(
+  previous: null | T,
+  processRunId: string,
+  active: { has(attemptId: string): boolean },
+): previous is T {
+  return (
+    previous?.state === 'running' &&
+    previous.processRunId === processRunId &&
+    active.has(previous.attemptId)
+  );
+}
+
 @Injectable()
 export class TbankAccessService {
   constructor(
@@ -40,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 },
       });
@@ -62,11 +76,7 @@ export class TbankAccessService {
       const previous = await tx.tbankSyncWork.findUnique({
         where: { userId_revision_scopeKey: scope },
       });
-      if (
-        previous?.state === 'running' &&
-        previous.processRunId === processRunId &&
-        active.has(previous.attemptId)
-      ) {
+      if (isJoinableRunningAttempt(previous, processRunId, active)) {
         return { work: previous, launch: false };
       }
       const data = {
@@ -98,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> {
@@ -151,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 },
@@ -232,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: {

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<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;
+}

src/integrations/tbank/tbank.api.ts

@@ -3,6 +3,7 @@ export { InstrumentMetadataService } from './instrument-metadata.service';
 export { metadataKey } from './instrument-metadata.service';
 export type { MetadataIdentity } from './instrument-metadata.service';
 export {
+  isJoinableRunningAttempt,
   TbankAccessService,
   type TbankSyncAttempt,
 } from './tbank-access.service';

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 },

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,

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(
@@ -110,6 +90,7 @@ export class CatalogEnrichmentService {
       tickers?: Record<string, string>;
       to: string;
     },
+    refreshBefore?: Date,
   ) {
     const { revision } = await this.access.local(userId);
     const inputs = sourceInputs(instrumentType, uids, kinds, range);
@@ -123,6 +104,7 @@ export class CatalogEnrichmentService {
         false,
         kinds,
         range,
+        refreshBefore,
       );
     }
     const work = inputs.length
@@ -175,6 +157,7 @@ export class CatalogEnrichmentService {
       tickers?: Record<string, string>;
       to: string;
     },
+    refreshBefore?: Date,
   ) {
     const inputs = sourceInputs(instrumentType, uids, kinds, range);
     const reservation = this.reservations.then(async () => {
@@ -189,7 +172,9 @@ export class CatalogEnrichmentService {
         );
         const policy = sourcePolicy(input.kind);
         const needsRefresh = force || !row?.fetchedAt ||
-          (policy === 'quote' && tbankSourceNeedsRefresh('quote', row.fetchedAt, new Date()));
+          (policy === 'quote' && tbankSourceNeedsRefresh('quote', row.fetchedAt, new Date())) ||
+          // A walk reads a saved calendar once: later accounts see the new fetchedAt.
+          (policy === 'calendar' && refreshBefore !== undefined && row.fetchedAt < refreshBefore);
         if (!needsRefresh) {
           continue;
         }
@@ -332,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 {
@@ -532,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;
@@ -634,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 (
@@ -665,6 +469,7 @@ function sourcePolicy(kind: CatalogSource['kind']) {
   ) {
     return 'quote';
   }
+  if (['bond-events', 'dividends'].includes(kind)) return 'calendar';
   return 'stable';
 }
 function producerKey(input: CatalogSource) {
@@ -673,6 +478,3 @@ function producerKey(input: CatalogSource) {
     ? `bond-metadata:${input.uid}`
     : catalogScope(input);
 }
-
-const quotationScale = 9;
-const quotationUnits = 1_000_000_000n;

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

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

src/modules/investments/instruments/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 },
       });

src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts

@@ -3,9 +3,10 @@ import { randomUUID } from 'node:crypto';
 import { isDeepStrictEqual } from 'node:util';
 
 import type { Prisma } from '../../../../../generated/prisma/client';
-import type { TbankBond, TbankShare, TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
 
 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';
 
 const UNCHANGED_REVISION_INCREMENT = 0;
@@ -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,20 +279,13 @@ 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 },
       });
-      if (
-        previous?.state === 'running' &&
-        previous.processRunId === runId &&
-        active.has(previous.attemptId)
-      ) {
+      if (isJoinableRunningAttempt(previous, runId, active)) {
         return { work: previous, launch: false };
       }
       const data = {
@@ -345,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 };
@@ -465,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,
@@ -473,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,
       });
     });
   }

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,
   ],

src/modules/investments/instruments/instruments.module.ts

@@ -1,6 +1,5 @@
 import { Module } from '@nestjs/common';
 
-import { TbankAccessService } from '../../../integrations/tbank/tbank-access.service';
 import { TbankModule } from '../../../integrations/tbank/tbank.module';
 import { AuthModule } from '../../auth/auth.module';
 import { PortfolioModule } from '../portfolio/portfolio.module';
@@ -21,7 +20,6 @@ import { StockController } from './stocks/presentation/stock.controller';
   controllers: [StockController, BondController],
   exports: [StockIncomeService, BondIncomeService],
   providers: [
-    { provide: 'TBANK_ACCESS', useExisting: TbankAccessService },
     BondIncomeService,
     BondMarketService,
     GetStockCompanyUseCase,

src/modules/investments/instruments/stocks/application/stock-screener.service.spec.ts

@@ -1,3 +1,7 @@
+jest.mock('../../../../../integrations/tbank/tbank.api', () => ({
+  TbankAccessService: class {},
+}));
+
 import { NotFoundException, ServiceUnavailableException } from '@nestjs/common';
 
 import type { TbankShare } from '../../../../../integrations/tbank/tbank.api';
@@ -36,7 +40,7 @@ describe('StockScreenerService', () => {
     client.getAllShares.mockResolvedValue(
       Array.from({ length: pageSize + 1 }, (_, index) => share(index)),
     );
-    service = new StockScreenerService(access, client as never, logger as never);
+    service = new StockScreenerService(access as never, client as never, logger as never);
   });
 
   it('stores the full selected catalog and returns stable non-overlapping pages', async () => {

src/modules/investments/instruments/stocks/application/stock-screener.service.ts

@@ -1,6 +1,5 @@
 import {
   BadRequestException,
-  Inject,
   Injectable,
   NotFoundException,
   ServiceUnavailableException,
@@ -10,6 +9,7 @@ import { randomUUID } from 'node:crypto';
 import type { TbankShare } from '../../../../../integrations/tbank/tbank-market.types';
 
 import { StructuredLogger } from '../../../../../common/logging/structured-logger.service';
+import { TbankAccessService } from '../../../../../integrations/tbank/tbank.api';
 import { TbankClient } from '../../../../../integrations/tbank/tbank.client';
 import {
   selectStockScreenerInstruments,
@@ -47,17 +47,13 @@ type StockScreenerPage = {
   searchId: string;
 };
 type Snapshot = { asOf: string; expiresAt: number; items: StockScreenerItem[]; revision: string; userId: string };
-type TbankAccess = {
-  local(userId: string): Promise<{ revision: string }>;
-  online<T>(userId: string, run: (connection: { revision: string; token: string }) => Promise<T>): Promise<T>;
-};
 
 @Injectable()
 export class StockScreenerService {
   private readonly snapshots = new Map<string, Snapshot>();
 
   constructor(
-    @Inject('TBANK_ACCESS') private readonly access: TbankAccess,
+    private readonly access: TbankAccessService,
     private readonly client: TbankClient,
     private readonly logger: StructuredLogger,
   ) {}

src/modules/investments/instruments/stocks/presentation/stock.controller.ts

@@ -91,6 +91,7 @@ export class StockController {
     @Body(new ValidationPipe({ forbidNonWhitelisted: true, forbidUnknownValues: true, whitelist: true }))
     criteria: StockScreenerCriteriaDto,
   ) {
+    // The global whitelist pipe strips unknown keys before this param pipe runs, so only the raw body can reject them.
     const allowed = new Set(['countryOfRisk', 'currency', 'maxLot', 'query', 'realExchange', 'sector', 'shareType']);
     if (Object.keys(request.body as object).some((key) => !allowed.has(key))) {
       throw new BadRequestException();

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<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;
 }
 
 function decodeInterval(key: string): OperationInterval {

src/modules/investments/portfolio/application/forecast.service.ts

@@ -20,6 +20,7 @@ export class ForecastService {
     lastAttempt?: InvestmentResourceMetaDto['lastAttempt'];
     membershipRevision: number;
     portfolio: InvestmentResourceMetaDto;
+    replacesLatest?: boolean;
     revision: string;
     scopeKey: string;
     sources: {
@@ -74,7 +75,7 @@ export class ForecastService {
           .map((dependency) => dependency.sync)
           .find((sync) => sync.state === 'failed');
     return {
-      data: stored ? filterEvents(stored.payload as T, input.scopeKey) : null,
+      data: stored ? filterEvents(stored.payload as T, input.scopeKey, stored.scopeKey) : null,
       meta: investmentResourceMeta({
         resourceKey: input.scopeKey,
         dataRevision: String(stored?.dataRevision ?? MISSING_DATA_REVISION),
@@ -95,7 +96,7 @@ export class ForecastService {
   }
 }
 
-function filterEvents<T extends object>(payload: T, scopeKey: string): T {
+function filterEvents<T extends object>(payload: T, scopeKey: string, savedScopeKey: string): T {
   if (!scopeKey.startsWith('events/')) return payload;
   try {
     const [, from, to] = JSON.parse(scopeKey.slice('events/'.length)) as [string, string, string];
@@ -111,8 +112,10 @@ function filterEvents<T extends object>(payload: T, scopeKey: string): T {
       const current = totals.get(event.projectedAmount.currency) ?? { value: 0n, scale: 0 };
       totals.set(event.projectedAmount.currency, add(current, decimalFromString(event.projectedAmount.value)));
     }
+    const [, savedFrom, savedTo] = JSON.parse(savedScopeKey.slice('events/'.length)) as [string, string, string];
     return {
       ...value,
+      window: { from: savedFrom, to: savedTo },
       events,
       incomeTotalsByCurrency: [...totals]
         .sort(([left], [right]) => left.localeCompare(right))

src/modules/investments/portfolio/application/portfolio.service.spec.ts

@@ -218,6 +218,18 @@ describe('persistent portfolio projection', () => {
     expect(snapshot.structure.currencies[0].totalValue).toBe('121.49');
   });
 
+  // Test intent: a corrupted Json payload must fail loudly instead of being spread into the API response.
+  it('rejects a stored snapshot payload without the persisted shape', async () => {
+    repository.read.mockResolvedValue({
+      membershipAt,
+      membershipRevision: 1,
+      accounts: [{ accountId: 'a', name: 'A', type: 'BROKERAGE', openedOn: null,
+        portfolio: { payload: { positions: 'broken' }, fetchedAt: membershipAt, dataRevision: 1 } }],
+      work: [],
+    });
+    await expect(service.read('user', 'all')).rejects.toThrow('Corrupted portfolio snapshot payload');
+  });
+
   it('does not turn a missing account snapshot into aggregate zero; summaries stay independent', async () => {
     const snapshot: PortfolioDto = {
       lastSuccessfulAt: membershipAt.toISOString(),

src/modules/investments/portfolio/application/portfolio.service.ts

@@ -4,6 +4,8 @@ import {
 } from '@nestjs/common';
 import { randomUUID } from 'node:crypto';
 
+import type { Prisma } from '../../../../../generated/prisma/client';
+
 import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
 import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
 import {
@@ -26,6 +28,7 @@ import { sumMoneyByCurrency } from '../domain/portfolio-totals';
 import {
   portfolioFreshForMs,
   PortfolioRepository,
+  portfolioSnapshot,
   portfolioWorkKey,
 } from '../infrastructure/portfolio.repository';
 import { mapPortfolio } from '../infrastructure/tbank-portfolio.mapper';
@@ -88,18 +91,7 @@ export class PortfolioService {
           type: account.type,
           openedOn: account.openedOn,
           snapshot: account.portfolio
-            ? {
-                ...(account.portfolio.payload as unknown as PortfolioDto),
-                positions: (account.portfolio.payload as unknown as PortfolioDto).positions.map((position) => ({
-                  ...position,
-                  sector: position.sector ?? null,
-                  issuer: position.issuer ?? null,
-                })),
-                lastSuccessfulAt: account.portfolio.fetchedAt.toISOString(),
-                isStale:
-                  Date.now() - account.portfolio.fetchedAt.valueOf() >=
-                  portfolioFreshForMs,
-              }
+            ? savedSnapshot(account.portfolio)
             : null,
           meta: investmentResourceMeta({
             resourceKey: `portfolio/${encodeURIComponent(account.accountId)}`,
@@ -326,3 +318,17 @@ export class PortfolioService {
     return mapPortfolio(upstream, money, savedMetadata, lastSuccessfulAt, classifications);
   }
 }
+
+function savedSnapshot(saved: { fetchedAt: Date; payload: Prisma.JsonValue }): PortfolioDto {
+  const snapshot = portfolioSnapshot(saved.payload);
+  return {
+    ...snapshot,
+    positions: snapshot.positions.map((position) => ({
+      ...position,
+      sector: position.sector ?? null,
+      issuer: position.issuer ?? null,
+    })),
+    lastSuccessfulAt: saved.fetchedAt.toISOString(),
+    isStale: Date.now() - saved.fetchedAt.valueOf() >= portfolioFreshForMs,
+  };
+}

src/modules/investments/portfolio/infrastructure/forecast.repository.ts

@@ -26,6 +26,7 @@ export class ForecastRepository {
     membershipRevision: number;
     portfolioProjection: string;
     portfolios: { accountId: string; dataRevision: number }[];
+    replacesLatest?: boolean;
     revision: string;
     scopeKey: string;
     sources: ForecastSourceVersion[];
@@ -51,32 +52,30 @@ export class ForecastRepository {
       const previous = await tx.tbankForecast.findUnique({
         where: { userId_revision_scopeKey: key },
       });
+      // An events read serves the account's latest complete forecast whatever
+      // the requested window; other scopes keep their exact-key row.
       const requestedWindow = eventsWindow(input.scopeKey);
-      const saved = previous ?? (requestedWindow
-        ? (
-            await tx.tbankForecast.findMany({
-              where: {
-                ...owner,
-                scopeKey: { startsWith: 'events/' },
-                membershipRevision: input.membershipRevision,
+      const latest = requestedWindow
+        ? await tx.tbankForecast.findFirst({
+            where: {
+              ...owner,
+              // scopeKey is `events/[accountId,from,to]` JSON: pin the account.
+              scopeKey: {
+                startsWith: `events/[${JSON.stringify(requestedWindow.accountId)},`,
               },
-            })
-          ).find((forecast) => {
-            const window = eventsWindow(forecast.scopeKey);
-            return (
-              window &&
-              window.accountId === requestedWindow.accountId &&
-              window.from <= requestedWindow.to &&
-              requestedWindow.from <= window.to
-            );
+              membershipRevision: input.membershipRevision,
+            },
+            orderBy: [{ calculatedAt: 'desc' }, { scopeKey: 'asc' }],
           })
-        : null);
+        : previous;
       const fallback =
-        saved?.membershipRevision === input.membershipRevision
-          ? saved
-          : null;
+        latest?.membershipRevision === input.membershipRevision ? latest : null;
       if (!input.candidate) return fallback;
-      if (!previous && fallback) return fallback;
+      // Only the default window may replace the latest forecast; an explicit
+      // range is served from it and publishes only when nothing is saved.
+      if (!input.replacesLatest && fallback && fallback.scopeKey !== input.scopeKey) {
+        return fallback;
+      }
       const portfolios = await tx.tbankPortfolio.findMany({
         where: {
           ...owner,
@@ -152,7 +151,7 @@ export class ForecastRepository {
       ) as Prisma.InputJsonObject;
       const timestamps = input.timestamps as Prisma.InputJsonArray;
       if (
-        fallback &&
+        fallback?.scopeKey === input.scopeKey &&
         isDeepStrictEqual(fallback.dependencyRevisions, revisions) &&
         isDeepStrictEqual(fallback.sourceTimestamps, timestamps)
       ) {

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 = {
@@ -200,3 +191,16 @@ export class PortfolioRepository {
 export function portfolioWorkKey(accountId: string): string {
   return `portfolio:${JSON.stringify(accountId)}`;
 }
+
+// Json payload is written only by publish(); a shape check turns corruption into an explicit 500.
+export function portfolioSnapshot(value: Prisma.JsonValue): PortfolioDto {
+  const snapshot = value as null | Partial<PortfolioDto>;
+  if (
+    !Array.isArray(snapshot?.positions) ||
+    !Array.isArray(snapshot.totalsByCurrency) ||
+    !Array.isArray(snapshot.cashBalances)
+  ) {
+    throw new Error('Corrupted portfolio snapshot payload');
+  }
+  return snapshot as PortfolioDto;
+}

src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts

@@ -268,6 +268,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
         account.accountId,
         dispatch,
         dispatchProblems,
+        walkStartedAt,
       );
     }
   }
@@ -340,9 +341,12 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
     accountId: string,
     dispatch: DispatchStep,
     dispatchProblems: HourlySection[],
+    walkStartedAt: Date,
   ) {
     try {
-      const ids = await this.withBound(this.events.refresh(userId, accountId));
+      const ids = await this.withBound(
+        this.events.refresh(userId, accountId, {}, walkStartedAt),
+      );
       if (ids !== BOUND_EXPIRED) {
         for (const id of ids) {
           await dispatch({ accountId, id, section: 'events' });

src/modules/investments/tbank-sync/tbank-sync.repository.spec.ts

@@ -0,0 +1,14 @@
+jest.mock('../../../prisma/prisma.service', () => ({
+  PrismaService: class {},
+}));
+
+import { screenIntent } from './tbank-sync.repository';
+
+describe('screen sync intent', () => {
+  it('restores a persisted intent and rejects a corrupted one', () => {
+    const intent = { now: '2026-11-02T00:00:00.000Z', scope: { screen: 'accounts' } };
+    expect(screenIntent(intent)).toEqual(intent);
+    expect(() => screenIntent({ source: 'hourly' })).toThrow('Corrupted screen sync intent');
+    expect(() => screenIntent(null)).toThrow('Corrupted screen sync intent');
+  });
+});

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

@@ -6,8 +6,18 @@ 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 };
+
+// Json columns are written only by this repository; a shape check turns corruption into an explicit 500.
+export function screenIntent(value: Prisma.JsonValue): ScreenIntent {
+  const intent = value as null | Partial<ScreenIntent>;
+  if (typeof intent?.now !== 'string' || typeof intent.scope?.screen !== 'string') {
+    throw new Error('Corrupted screen sync intent');
+  }
+  return intent as ScreenIntent;
+}
 export type ScreenProgress = {
   attempts?: Record<string, string>;
   children: string[];
@@ -79,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({
@@ -167,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 };
     });
   }
 
@@ -226,10 +227,8 @@ export class TbankSyncRepository {
         processRunId: runId,
         state: 'running',
         priority: 0,
-        intent: JSON.parse(
-          JSON.stringify({ source: 'hourly' }),
-        ) as Prisma.InputJsonObject,
-        progress: { sections: [] } as unknown as Prisma.InputJsonValue,
+        intent: { source: 'hourly' },
+        progress: { sections: [] },
         errorCode: null,
       };
       const work = await tx.tbankSyncWork.upsert({
@@ -357,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);
     });
   }
 

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

@@ -8,10 +8,7 @@ import { randomUUID } from 'node:crypto';
 import { TbankAccessService } from '../../../integrations/tbank/tbank.api';
 import { BrokerageOperationsSyncService } from '../operations/operations.api';
 import { TbankSyncStatusDto } from './tbank-sync.dto';
-import {
-  type ScreenIntent,
-  TbankSyncRepository,
-} from './tbank-sync.repository';
+import { screenIntent, TbankSyncRepository } from './tbank-sync.repository';
 
 @Injectable()
 export class TbankSyncService {
@@ -35,7 +32,7 @@ export class TbankSyncService {
     if (!saved.work.scopeKey.startsWith('screen:')) {
       return this.sourceStatus(userId, saved.work);
     }
-    const intent = saved.work.intent as unknown as ScreenIntent;
+    const intent = screenIntent(saved.work.intent);
     await this.access.local(userId, intent.scope.accountId);
     const children = await Promise.all(
       saved.children.map((child) =>

src/modules/investments/upcoming-events/upcoming-events.dto.ts

@@ -65,6 +65,14 @@ export class UpcomingEventDto {
   declare projectedAmount: MoneyDto | null;
 }
 
+export class UpcomingEventsWindowDto {
+  @ApiProperty({ format: 'date' })
+  declare from: string;
+
+  @ApiProperty({ format: 'date' })
+  declare to: string;
+}
+
 export class UpcomingEventsDto {
   @ApiProperty()
   declare accountId: string;
@@ -72,6 +80,13 @@ export class UpcomingEventsDto {
   @ApiProperty({ format: 'date-time' })
   declare calculatedAt: string;
 
+  @ApiProperty({
+    type: UpcomingEventsWindowDto,
+    description:
+      'Inclusive UTC window the saved forecast was calculated for. Unchanged by the requested from/to filter.',
+  })
+  declare window: UpcomingEventsWindowDto;
+
   @ApiProperty({ type: UpcomingEventDto, isArray: true })
   declare events: UpcomingEventDto[];
 

src/modules/investments/upcoming-events/upcoming-events.service.ts

@@ -150,7 +150,7 @@ export class UpcomingEventsService {
         a.type.localeCompare(b.type) ||
         a.ticker.localeCompare(b.ticker),
     );
-    const candidate: undefined | UpcomingEventsDto = ready
+    const candidate: Omit<UpcomingEventsDto, 'window'> | undefined = ready
       ? {
           accountId,
           calculatedAt: new Date().toISOString(),
@@ -163,6 +163,7 @@ export class UpcomingEventsService {
       revision: connection.revision,
       membershipRevision: connection.membershipRevision,
       scopeKey: `events/${JSON.stringify([accountId, window.from, window.to])}`,
+      replacesLatest: query.from === undefined && query.to === undefined,
       portfolio: portfolio.meta,
       lastAttempt: hourlyAttempt(
         hourly,
@@ -184,6 +185,7 @@ export class UpcomingEventsService {
     userId: string,
     accountId: string,
     query: UpcomingEventsQueryDto = {},
+    walkStartedAt?: Date,
   ): Promise<string[]> {
     const window = dateWindow(query);
     await this.access.local(userId, accountId);
@@ -194,6 +196,7 @@ export class UpcomingEventsService {
       positions,
       window,
       false,
+      walkStartedAt,
     );
     return launched;
   }
@@ -203,6 +206,7 @@ export class UpcomingEventsService {
     positions: Position[],
     window: { from: string; to: string },
     observe: boolean,
+    refreshBefore?: Date,
   ) {
     const metadata = await this.metadata.savedMany(positions);
     const sources: Awaited<ReturnType<CatalogEnrichmentService['readMany']>> =
@@ -228,6 +232,7 @@ export class UpcomingEventsService {
               identities.map((entry) => [entry.uid, entry.ticker]),
             ),
           },
+          refreshBefore,
         );
         sources.push(...batch);
         for (const source of batch) {

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

test/tbank-membership-reservation.e2e-spec.ts

@@ -0,0 +1,107 @@
+import { ConfigService } from '@nestjs/config';
+
+import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service';
+import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
+import { PrismaService } from '../src/prisma/prisma.service';
+
+// Characterizes the membership reserve protocol on real SQLite transactions.
+// Join by the same process while the attempt is active; launch after it is not.
+describe('membership reservation protocol (SQLite)', () => {
+  const prisma = new PrismaService(
+    new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
+  );
+  const tokens = new TbankTokenService(
+    new ConfigService({
+      TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY,
+    }) as never,
+  );
+  const access = new TbankAccessService(prisma, tokens);
+  const userId = 'synthetic-membership-reservation';
+  const revision = `${userId}-revision`;
+
+  beforeAll(async () => {
+    await prisma.$connect();
+    await prisma.tbankSyncWork.deleteMany({ where: { userId } });
+    await prisma.tbankConnection.deleteMany({ where: { userId } });
+    await prisma.user.deleteMany({ where: { id: userId } });
+    await prisma.user.create({
+      data: {
+        id: userId,
+        email: `${userId}@example.invalid`,
+        passwordHash: 'synthetic',
+      },
+    });
+    await prisma.tbankConnection.create({
+      data: {
+        userId,
+        revision,
+        encryptedToken: tokens.encrypt('synthetic-membership'),
+        connectedAt: new Date(),
+      },
+    });
+  });
+
+  afterAll(async () => {
+    await prisma.tbankSyncWork.deleteMany({ where: { userId } });
+    await prisma.tbankConnection.deleteMany({ where: { userId } });
+    await prisma.user.deleteMany({ where: { id: userId } });
+    await prisma.$disconnect();
+  });
+
+  it('launches once, joins the active attempt by the same process, and relaunches after it finishes', async () => {
+    const processRunId = 'run-membership-a';
+    const first = await access.reserveMembership(
+      userId,
+      revision,
+      processRunId,
+      new Set(),
+      true,
+    );
+    expect(first?.launch).toBe(true);
+    const attemptId = first?.work.attemptId ?? '';
+
+    const joined = await access.reserveMembership(
+      userId,
+      revision,
+      processRunId,
+      new Set([attemptId]),
+      true,
+    );
+    expect(joined?.launch).toBe(false);
+    expect(joined?.work.id).toBe(first?.work.id);
+    expect(joined?.work.attemptId).toBe(attemptId);
+
+    // The attempt is no longer in the active set: this must not join it.
+    const relaunched = await access.reserveMembership(
+      userId,
+      revision,
+      processRunId,
+      new Set(),
+      true,
+    );
+    expect(relaunched?.launch).toBe(true);
+    expect(relaunched?.work.attemptId).not.toBe(attemptId);
+  });
+
+  it('does not join an attempt started by another process', async () => {
+    const other = await access.reserveMembership(
+      userId,
+      revision,
+      'run-membership-b',
+      new Set(),
+      true,
+    );
+    expect(other?.launch).toBe(true);
+    const attemptId = other?.work.attemptId ?? '';
+
+    const foreign = await access.reserveMembership(
+      userId,
+      revision,
+      'run-membership-c',
+      new Set([attemptId]),
+      true,
+    );
+    expect(foreign?.launch).toBe(true);
+    expect(foreign?.work.attemptId).not.toBe(attemptId);
+  });
+});

test/upcoming-events.e2e-spec.ts

@@ -82,7 +82,10 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
       `brokerage-accounts/${scope}/events?from=${range.from}&to=${range.to}${observe ? '&sync=observe' : ''}`,
       userId,
     );
-  const refresh = async (range = window, accountId = 'account') => {
+  const refresh = async (
+    range: { from?: string; to?: string } = window,
+    accountId = 'account',
+  ) => {
     await app.get(UpcomingEventsService).refresh('owner', accountId, range);
     await settled();
   };
@@ -318,17 +321,23 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
     expect(upstream.getDividends).toHaveBeenCalledTimes(count);
   });
 
-  it('never relabels another range or membership snapshot and rejects inaccessible accounts locally', async () => {
+  it('serves a disjoint range as an empty saved result and never relabels a membership snapshot', async () => {
     await refresh();
     await settled();
-    await events().expect(200);
+    const saved = (await events().expect(200))
+      .body as UpcomingEventsResourceDto;
     const newRange = (
       await events(true, 'account', 'owner', {
         from: '2099-03-01',
         to: '2099-03-31',
       }).expect(200)
     ).body as UpcomingEventsResourceDto;
-    expect(newRange.data).toBeNull();
+    expect(newRange.data).toEqual({
+      ...saved.data,
+      events: [],
+      incomeTotalsByCurrency: [],
+    });
+    expect(newRange.data?.window).toEqual(window);
     await prisma.tbankConnection.update({
       where: { userId: 'owner' },
       data: { membershipRevision: { increment: 1 } },
@@ -640,6 +649,247 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
     expect(moex.bondEvents).not.toHaveBeenCalled();
   });
 
+  it('keeps the latest wide forecast when an explicit narrower range has a complete candidate', async () => {
+    await refresh();
+    const full = (await events().expect(200))
+      .body as UpcomingEventsResourceDto;
+    const narrowRange = { from: '2099-02-01', to: '2099-02-10' };
+    await refresh(narrowRange);
+
+    const narrow = (await events(true, 'account', 'owner', narrowRange).expect(
+      200,
+    )).body as UpcomingEventsResourceDto;
+
+    expect(narrow.data?.calculatedAt).toBe(full.data?.calculatedAt);
+    expect(narrow.data?.window).toEqual(window);
+    expect(narrow.data?.events.map((event) => event.date)).toEqual([
+      '2099-02-01',
+    ]);
+    expect(await prisma.tbankForecast.count()).toBe(1);
+  });
+
+  describe('daily default window shift', () => {
+    const day0 = '2099-02-01';
+    const day1 = '2099-02-02';
+    const windowOf = (from: string, to: string) => ({ from, to });
+    const w0 = windowOf(day0, '2099-03-03');
+    const w1 = windowOf(day1, '2099-03-04');
+    const dividend = (date: string) => ({
+      recordDate: time(date),
+      dividendNet: { units: '10', nano: 1, currency: 'usd' },
+    });
+    const defaultEvents = async () =>
+      (await get('brokerage-accounts/account/events?sync=observe').expect(200))
+        .body as UpcomingEventsResourceDto;
+    // Only Date is faked: the app and SQLite helpers need real timers.
+    const clock = (date: string) => {
+      jest.useFakeTimers({
+        doNotFake: [
+          'hrtime', 'nextTick', 'performance', 'queueMicrotask',
+          'requestAnimationFrame', 'cancelAnimationFrame',
+          'requestIdleCallback', 'cancelIdleCallback', 'setImmediate',
+          'clearImmediate', 'setInterval', 'clearInterval', 'setTimeout',
+          'clearTimeout',
+        ],
+        now: new Date(`${date}T12:00:00Z`),
+      });
+    };
+    afterEach(() => jest.useRealTimers());
+
+    it('publishes the new window as latest and serves it for stale or disjoint explicit ranges', async () => {
+      clock(day0);
+      upstream.getDividends.mockResolvedValue([
+        dividend('2099-02-05'),
+        dividend('2099-03-04'),
+      ]);
+      await portfolio([position(shareUid, '2')]);
+      await refresh({});
+      const first = await defaultEvents();
+      expect(first.data?.window).toEqual(w0);
+      expect(first.data?.incomeTotalsByCurrency).toEqual([
+        { currency: 'USD', value: '20.000000002' },
+      ]);
+
+      jest.setSystemTime(new Date(`${day1}T12:00:00Z`));
+      await portfolio([position(shareUid, '5')]);
+      await refresh({});
+      const second = await defaultEvents();
+
+      expect(second.data?.window).toEqual(w1);
+      expect(second.data?.calculatedAt).not.toBe(first.data?.calculatedAt);
+      expect(second.data?.events.map((event) => event.date)).toEqual([
+        '2099-02-05',
+        '2099-03-04',
+      ]);
+      expect(second.data?.incomeTotalsByCurrency).toEqual([
+        { currency: 'USD', value: '100.00000001' },
+      ]);
+      const stale = (await events(true, 'account', 'owner', w0).expect(200))
+        .body as UpcomingEventsResourceDto;
+      expect(stale.data?.calculatedAt).toBe(second.data?.calculatedAt);
+      expect(stale.data?.window).toEqual(w1);
+      const disjoint = (
+        await events(true, 'account', 'owner', {
+          from: '2100-01-01',
+          to: '2100-01-31',
+        }).expect(200)
+      ).body as UpcomingEventsResourceDto;
+      expect(disjoint.data).toEqual({
+        ...second.data,
+        events: [],
+        incomeTotalsByCurrency: [],
+      });
+      expect(await prisma.tbankForecast.count()).toBe(2);
+    });
+
+    it('keeps the previous complete forecast when a holding calendar fails after the shift', async () => {
+      clock(day0);
+      upstream.getDividends.mockResolvedValue([dividend('2099-02-05')]);
+      await refresh({});
+      const before = await defaultEvents();
+      expect(before.data?.window).toEqual(w0);
+
+      jest.setSystemTime(new Date(`${day1}T12:00:00Z`));
+      await portfolio([position(shareUid, '7'), position(newUid, '5')]);
+      upstream.getDividends.mockImplementation((_token: string, uid: string) =>
+        uid === newUid
+          ? Promise.reject(new Error('new calendar unavailable'))
+          : Promise.resolve([dividend('2099-02-05')]),
+      );
+      await refresh({});
+
+      expect((await defaultEvents()).data).toEqual(before.data);
+      expect(await prisma.tbankForecast.count()).toBe(1);
+    });
+
+    it('publishes the shifted window even when no source depends on it', async () => {
+      clock(day0);
+      await portfolio([]);
+      await refresh({});
+      expect((await defaultEvents()).data?.window).toEqual(w0);
+
+      jest.setSystemTime(new Date(`${day1}T12:00:00Z`));
+      await refresh({});
+
+      expect((await defaultEvents()).data?.window).toEqual(w1);
+    });
+  });
+
+  describe('hourly pass refreshes saved payout calendars', () => {
+    const date = new Date(Date.now() + 7 * 86400000).toISOString().slice(0, 10);
+    const dividend = (units: string, dividendType = 'Regular') => ({
+      recordDate: time(date),
+      dividendType,
+      dividendNet: { units, currency: 'usd' },
+    });
+    // One hourly walk: the scheduler stamps it once and passes it to every account.
+    const startWalk = async () => {
+      await delay(10);
+      return new Date();
+    };
+    const pass = async (walkStartedAt: Date, accountId = 'account') => {
+      await app
+        .get(UpcomingEventsService)
+        .refresh('owner', accountId, {}, walkStartedAt);
+      await settled();
+    };
+    const latest = async () =>
+      (await get('brokerage-accounts/account/events?sync=observe').expect(200))
+        .body as UpcomingEventsResourceDto;
+    const dividendRow = () =>
+      prisma.tbankInstrumentSource.findFirstOrThrow({
+        where: { instrumentUid: shareUid, kind: 'dividends' },
+      });
+
+    it('picks up a changed and then a cancelled payout on the next pass of the same window', async () => {
+      upstream.getDividends.mockResolvedValue([dividend('10')]);
+      await pass(await startWalk());
+      expect((await latest()).data?.incomeTotalsByCurrency).toEqual([
+        { currency: 'USD', value: '20' },
+      ]);
+      const first = await dividendRow();
+
+      upstream.getDividends.mockResolvedValue([dividend('30')]);
+      await pass(await startWalk());
+      expect((await latest()).data?.incomeTotalsByCurrency).toEqual([
+        { currency: 'USD', value: '60' },
+      ]);
+      expect((await dividendRow()).dataRevision).toBe(first.dataRevision + 1);
+
+      upstream.getDividends.mockResolvedValue([dividend('30', 'Cancelled')]);
+      await pass(await startWalk());
+      expect((await latest()).data?.events).toEqual([]);
+    });
+
+    it('keeps the first complete calendar, time and forecast when the next refresh fails', async () => {
+      upstream.getDividends.mockResolvedValue([dividend('10')]);
+      await pass(await startWalk());
+      const before = await latest();
+      const row = await dividendRow();
+
+      upstream.getDividends.mockRejectedValue(new Error('outage'));
+      await pass(await startWalk());
+      const after = await latest();
+
+      expect(after.data).toEqual(before.data);
+      expect(after.meta.lastSuccessfulAt).toBe(before.meta.lastSuccessfulAt);
+      expect(
+        after.meta.dependencies?.some(
+          (dependency) => dependency.lastAttempt.state === 'failed',
+        ),
+      ).toBe(true);
+      const kept = await dividendRow();
+      expect(kept.dataRevision).toBe(row.dataRevision);
+      expect(kept.fetchedAt).toEqual(row.fetchedAt);
+    });
+
+    it('loads one shared calendar once per walk across accounts and leaves detail, GET and a plain refresh alone', async () => {
+      await prisma.tbankAccount.create({
+        data: {
+          userId: 'owner',
+          revision: 'owner-revision',
+          accountId: 'second',
+          type: 'ACCOUNT_TYPE_TINKOFF',
+          name: 'Second',
+          upstreamOrder: 1,
+        },
+      });
+      await portfolio([position()], 'owner', 'second');
+      upstream.getDividends.mockResolvedValue([dividend('10')]);
+      await pass(await startWalk());
+      const detailCalls = () =>
+        upstream.getShare.mock.calls.length +
+        upstream.getInstrumentByUid.mock.calls.length;
+      const detailBefore = detailCalls();
+      expect(upstream.getDividends).toHaveBeenCalledTimes(1);
+
+      const walk = await startWalk();
+      await pass(walk, 'account');
+      await pass(walk, 'second');
+      expect(upstream.getDividends).toHaveBeenCalledTimes(2);
+
+      await app.get(UpcomingEventsService).refresh('owner', 'account', {});
+      await settled();
+      await latest();
+      await get('brokerage-accounts/all/events').expect(200);
+      expect(upstream.getDividends).toHaveBeenCalledTimes(2);
+      expect(detailCalls()).toBe(detailBefore);
+    });
+
+    it('re-reads the bond calendar whose key never changes, only on a walk pass', async () => {
+      await portfolio([position(bondUid, '3', 'BOND')]);
+      await pass(await startWalk());
+      expect(moex.bondEvents).toHaveBeenCalledTimes(1);
+
+      await app.get(UpcomingEventsService).refresh('owner', 'account', {});
+      await settled();
+      expect(moex.bondEvents).toHaveBeenCalledTimes(1);
+
+      await pass(await startWalk());
+      expect(moex.bondEvents).toHaveBeenCalledTimes(2);
+    });
+  });
+
   it('discards a late calendar from a replaced connection and prevents access to old private forecasts', async () => {
     let release: (value: unknown[]) => void = () => {};
     upstream.getDividends.mockReturnValueOnce(