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

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

@@ -12,6 +12,7 @@ import type { TbankInstrument } from './tbank.client';
 import { PrismaService } from '../../prisma/prisma.service';
 import { isJoinableRunningAttempt, type TbankSyncAttempt } from './tbank-access.service';
 import { TbankErrorCode } from './tbank-error-code';
+import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim';
 
 const FIRST_BATCH_OFFSET = 0;
 const UNCHANGED_REVISION_INCREMENT = 0;
@@ -119,11 +120,8 @@ export class InstrumentMetadataRepository {
     active: ReadonlySet<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,

src/integrations/tbank/tbank-access.service.ts

@@ -12,6 +12,7 @@ import { PrismaService } from '../../prisma/prisma.service';
 import { supportedTbankAccounts } from './tbank-account.policy';
 import { TbankErrorCode } from './tbank-error-code';
 import { tbankSourceNeedsRefresh } from './tbank-source.policy';
+import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim';
 import { TbankTokenService } from './tbank-token.service';
 
 const EXPECTED_UPDATED_CONNECTIONS = 1;
@@ -56,11 +57,8 @@ export class TbankAccessService {
     force: boolean,
   ) {
     return this.prisma.$transaction(async (tx) => {
-      const current = await tx.tbankConnection.updateMany({
-        where: { userId, revision, invalidToken: false },
-        data: { membershipRevision: { increment: 0 } },
-      });
-      if (!current.count) return null;
+      const current = await lockConnectionGeneration(tx, { userId, revision });
+      if (!current) return null;
       const connection = await tx.tbankConnection.findUniqueOrThrow({
         where: { userId },
       });
@@ -110,15 +108,11 @@ export class TbankAccessService {
 
   async failMembership(attempt: TbankSyncAttempt, invalidToken: boolean) {
     if (invalidToken) return this.rejectMembershipToken(attempt);
-    const result = await this.prisma.tbankSyncWork.updateMany({
-      where: {
-        ...attempt,
-        state: 'running',
-        connection: { invalidToken: false },
-      },
-      data: { state: 'failed', errorCode: TbankErrorCode.Unavailable },
+    const result = await claimSyncWork(this.prisma, attempt, {
+      state: 'failed',
+      errorCode: TbankErrorCode.Unavailable,
     });
-    return result.count === EXPECTED_UPDATED_CONNECTIONS;
+    return result === EXPECTED_UPDATED_CONNECTIONS;
   }
 
   async tokenForAttempt(attempt: TbankSyncAttempt): Promise<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: {

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/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: {
+      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

@@ -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: {
+      await claimSyncWork(tx, attempt, {
         state: success ? 'complete' : 'failed',
         errorCode: success ? null : TbankErrorCode.Unavailable,
-        },
       });
     });
   }

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,
+      const locked = await lockConnectionGeneration(tx, { userId, revision }, {
         membershipAt: { not: null },
-        },
-        data: { membershipRevision: { increment: 0 } },
       });
-      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/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,
+      const current = await lockConnectionGeneration(tx, { userId, revision }, {
         accounts: { some: { accountId } },
-        },
-        data: { membershipRevision: { increment: 0 } },
       });
-      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 = {

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

@@ -6,6 +6,7 @@ import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api';
 import type { InvestmentAttemptDto } from '../investment-resource.dto';
 import type { InvestmentScreenDto } from './tbank-sync.dto';
 
+import { claimSyncWork, lockConnectionGeneration } from '../../../integrations/tbank/tbank-sync-claim';
 import { PrismaService } from '../../../prisma/prisma.service';
 export type ScreenIntent = { now: string; scope: InvestmentScreenDto };
 
@@ -88,11 +89,8 @@ export class TbankSyncRepository {
     runId: string,
   ) {
     return this.prisma.$transaction(async (tx) => {
-      const valid = await tx.tbankConnection.updateMany({
-        where: { userId, revision, invalidToken: false },
-        data: { membershipRevision: { increment: 0 } },
-      });
-      if (!valid.count) return null;
+      const valid = await lockConnectionGeneration(tx, { userId, revision });
+      if (!valid) return null;
       const scopeKey = `screen:${JSON.stringify(scope)}`;
       const key = { userId, revision, scopeKey };
       const previous = await tx.tbankSyncWork.findUnique({
@@ -176,17 +174,11 @@ export class TbankSyncRepository {
         ...Object.fromEntries(rows.map((row) => [row.id, row.attemptId])),
         ...old.attempts,
       };
-      return tx.tbankSyncWork.updateMany({
-        where: {
-          ...attempt,
-          state: 'running',
-          connection: { invalidToken: false },
-        },
-        data: {
+      const claimed = await claimSyncWork(tx, attempt, {
         progress: { children: [...new Set(children)], attempts, dispatched },
         ...(errorCode ? { state: 'failed', errorCode } : {}),
-        },
       });
+      return { count: claimed };
     });
   }
 
@@ -364,9 +356,7 @@ export class TbankSyncRepository {
         where: { userId: attempt.userId, id: child.id },
         select: { attemptId: true },
       });
-      const updated = await tx.tbankSyncWork.updateMany({
-        where: { ...attempt, state: 'running' },
-        data: {
+      const updated = await claimSyncWork(tx, attempt, {
         progress: {
           attempts: {
             ...progress.attempts,
@@ -375,9 +365,8 @@ export class TbankSyncRepository {
           children: [...(progress.children ?? []), child],
           sections: progress.sections ?? [],
         },
-        },
-      });
-      return Boolean(updated.count);
+      }, { withoutConnectionFence: true });
+      return Boolean(updated);
     });
   }