diff --git a/src/integrations/tbank/instrument-metadata.repository.ts b/src/integrations/tbank/instrument-metadata.repository.ts index 73a0396..8003192 100644 --- a/src/integrations/tbank/instrument-metadata.repository.ts +++ b/src/integrations/tbank/instrument-metadata.repository.ts @@ -12,6 +12,7 @@ import type { TbankInstrument } from './tbank.client'; import { PrismaService } from '../../prisma/prisma.service'; import { isJoinableRunningAttempt, type TbankSyncAttempt } from './tbank-access.service'; import { TbankErrorCode } from './tbank-error-code'; +import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim'; const FIRST_BATCH_OFFSET = 0; const UNCHANGED_REVISION_INCREMENT = 0; @@ -119,11 +120,8 @@ export class InstrumentMetadataRepository { active: ReadonlySet, ) { return this.prisma.$transaction(async (tx) => { - const valid = await tx.tbankConnection.updateMany({ - where: { ...attemptOwner, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!valid.count) return null; + const valid = await lockConnectionGeneration(tx, attemptOwner); + if (!valid) return null; const scope = { ...attemptOwner, scopeKey: `metadata:${metadataKey(identity)}`, @@ -167,15 +165,11 @@ export class InstrumentMetadataRepository { publish(attempt: TbankSyncAttempt, results: MetadataLookup[], now: Date) { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', progress: {} }, + const claimed = await claimSyncWork(tx, attempt, { + state: 'complete', + progress: {}, }); - if (!claimed.count) return false; + if (!claimed) return false; const aliases = new Map( results.map((result) => [`${result.type}:${result.id}`, result]), ); @@ -264,18 +258,17 @@ export class InstrumentMetadataRepository { } async finishObserver(attempt: TbankSyncAttempt, success: boolean) { - await this.prisma.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - ...(success ? { connection: { invalidToken: false } } : {}), - }, - data: { + await claimSyncWork( + this.prisma, + attempt, + { state: success ? 'complete' : 'failed', errorCode: success ? null : TbankErrorCode.Unavailable, }, - }); + { withoutConnectionFence: !success }, + ); if (success) { + // Fails sibling attempts whose connection was invalidated mid-flight. await this.prisma.tbankSyncWork.updateMany({ where: { ...attempt, diff --git a/src/integrations/tbank/tbank-access.service.ts b/src/integrations/tbank/tbank-access.service.ts index 98f030a..adcc5ed 100644 --- a/src/integrations/tbank/tbank-access.service.ts +++ b/src/integrations/tbank/tbank-access.service.ts @@ -12,6 +12,7 @@ import { PrismaService } from '../../prisma/prisma.service'; import { supportedTbankAccounts } from './tbank-account.policy'; import { TbankErrorCode } from './tbank-error-code'; import { tbankSourceNeedsRefresh } from './tbank-source.policy'; +import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim'; import { TbankTokenService } from './tbank-token.service'; const EXPECTED_UPDATED_CONNECTIONS = 1; @@ -56,11 +57,8 @@ export class TbankAccessService { force: boolean, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { userId, revision, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!current.count) return null; + const current = await lockConnectionGeneration(tx, { userId, revision }); + if (!current) return null; const connection = await tx.tbankConnection.findUniqueOrThrow({ where: { userId }, }); @@ -110,15 +108,11 @@ export class TbankAccessService { async failMembership(attempt: TbankSyncAttempt, invalidToken: boolean) { if (invalidToken) return this.rejectMembershipToken(attempt); - const result = await this.prisma.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'failed', errorCode: TbankErrorCode.Unavailable }, + const result = await claimSyncWork(this.prisma, attempt, { + state: 'failed', + errorCode: TbankErrorCode.Unavailable, }); - return result.count === EXPECTED_UPDATED_CONNECTIONS; + return result === EXPECTED_UPDATED_CONNECTIONS; } async tokenForAttempt(attempt: TbankSyncAttempt): Promise { @@ -163,15 +157,8 @@ export class TbankAccessService { return this.prisma.$transaction(async (tx) => { // Acquire the write fence before reading membership. Replacement and // attempt supersession cannot interleave with the publication below. - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { updatedAt: new Date() }, - }); - if (!claimed.count) return false; + const claimed = await claimSyncWork(tx, attempt, { updatedAt: new Date() }); + if (!claimed) return false; const { userId, revision } = attempt; const previous = await tx.tbankAccount.findMany({ where: { userId, revision }, @@ -244,11 +231,11 @@ export class TbankAccessService { async rejectMembershipToken(attempt: TbankSyncAttempt): Promise { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { ...attempt, state: 'running' }, - data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken }, - }); - if (!claimed.count) return false; + const claimed = await claimSyncWork(tx, attempt, { + state: 'failed', + errorCode: TbankErrorCode.InvalidToken, + }, { withoutConnectionFence: true }); + if (!claimed) return false; await tx.tbankConnection.update({ where: { userId_revision: { diff --git a/src/integrations/tbank/tbank-sync-claim.ts b/src/integrations/tbank/tbank-sync-claim.ts new file mode 100644 index 0000000..2445398 --- /dev/null +++ b/src/integrations/tbank/tbank-sync-claim.ts @@ -0,0 +1,53 @@ +import type { Prisma } from '../../../generated/prisma/client'; +import type { TbankSyncAttempt } from './tbank-access.service'; + +const NO_UPDATED_ROWS = 0; + +// Shared owners for the tbank-sync DB claim/lock protocol. +// Both helpers are CAS fences: they run inside the caller's transaction and +// keep the exact where/data shapes they replaced. Callers keep deciding +// success from the claim count exactly as before. + +// Claims the active TbankSyncWork row: only the still-running attempt of a +// valid connection may transition. `data` is the state transition to apply. +export type ClaimOptions = { + // Overrides the default 'running' fence (e.g. an initiator row also + // accepts 'complete'). + state: Prisma.TbankSyncWorkWhereInput['state'] | undefined; + // true drops the `connection: { invalidToken: false }` guard for fences + // that must also fire on an invalidated connection. + withoutConnectionFence: boolean | undefined; +}; +export async function claimSyncWork( + tx: Prisma.TransactionClient, + attempt: TbankSyncAttempt, + data: Prisma.TbankSyncWorkUpdateManyMutationInput, + options: Partial = {}, +): Promise { + const result = await tx.tbankSyncWork.updateMany({ + where: { + ...attempt, + state: options.state ?? 'running', + ...(options.withoutConnectionFence + ? {} + : { connection: { invalidToken: false } }), + }, + data, + }); + return result.count; +} + +// Locks the connection generation before reserving or publishing work: the +// row must still be the current revision with a valid token. `extra` +// (e.g. account membership) is merged into the where clause. +export async function lockConnectionGeneration( + tx: Prisma.TransactionClient, + owner: { revision: string; userId: string }, + extra: Prisma.TbankConnectionWhereInput = {}, +): Promise { + const result = await tx.tbankConnection.updateMany({ + where: { ...owner, ...extra, invalidToken: false }, + data: { membershipRevision: { increment: 0 } }, + }); + return result.count > NO_UPDATED_ROWS; +} diff --git a/src/modules/investments/instruments/infrastructure/candle.repository.ts b/src/modules/investments/instruments/infrastructure/candle.repository.ts index d01ec5b..7d308f3 100644 --- a/src/modules/investments/instruments/infrastructure/candle.repository.ts +++ b/src/modules/investments/instruments/infrastructure/candle.repository.ts @@ -2,8 +2,15 @@ import { Injectable } from '@nestjs/common'; import { createHash, randomUUID } from 'node:crypto'; import type { Prisma } from '../../../../../generated/prisma/client'; -import type { TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api'; +import { + claimSyncWork, + lockConnectionGeneration, +} from '../../../../integrations/tbank/tbank-sync-claim'; +import { + isJoinableRunningAttempt, + type TbankSyncAttempt, +} from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; import { type CandleRange, @@ -62,34 +69,30 @@ export class CandleRepository { priority = 1, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { ...owner, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!current.count) return null; + const current = await lockConnectionGeneration(tx, owner); + if (!current) return null; const scope = { ...owner, scopeKey }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: scope }, }); - if ( - previous?.state === 'running' && - previous.processRunId === runId && - active.has(previous.attemptId) - ) { - const previousIntent = previous.intent as Prisma.JsonObject; + const joinable = isJoinableRunningAttempt(previous, runId, active) + ? previous + : null; + if (joinable) { + const previousIntent = joinable.intent as Prisma.JsonObject; const previousNow = typeof previousIntent.now === 'string' ? previousIntent.now : ''; - const nextPriority = Math.min(previous.priority, priority); + const nextPriority = Math.min(joinable.priority, priority); if ( - nextPriority === previous.priority && + nextPriority === joinable.priority && (!intent.force || (previousIntent.force === true && (!intent.now || intent.now <= previousNow))) ) { - return { work: previous, launch: false }; + return { work: joinable, launch: false }; } const work = await tx.tbankSyncWork.update({ - where: { id: previous.id }, + where: { id: joinable.id }, data: { priority: nextPriority, intent: { @@ -163,15 +166,10 @@ export class CandleRepository { ).values(), ]; return this.prisma.$transaction(async (tx) => { - const valid = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { progress: { lastWindow: key } }, + const valid = await claimSyncWork(tx, attempt, { + progress: { lastWindow: key }, }); - if (!valid.count) return false; + if (!valid) return false; const identity = { workId: attempt.id, attemptId: attempt.attemptId, @@ -196,18 +194,11 @@ export class CandleRepository { completeWork = true, ) { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { - ...(completeWork ? { state: 'complete' } : {}), - errorCode: null, - }, + const claimed = await claimSyncWork(tx, attempt, { + ...(completeWork ? { state: 'complete' } : {}), + errorCode: null, }); - if (!claimed.count) return false; + if (!claimed) return false; const staged = await tx.tbankSyncStage.findMany({ where: { workId: attempt.id, @@ -311,28 +302,17 @@ export class CandleRepository { }, }); if (!origin) return false; - const result = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', errorCode: null }, + const result = await claimSyncWork(tx, attempt, { + state: 'complete', + errorCode: null, }); - return result.count > NO_UPDATED_ROWS; + return result > NO_UPDATED_ROWS; }); } fail(attempt: TbankSyncAttempt, errorCode: string) { return this.prisma.$transaction(async (tx) => { - await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'failed', errorCode }, - }); + await claimSyncWork(tx, attempt, { state: 'failed', errorCode }); await tx.tbankSyncStage.deleteMany({ where: { workId: attempt.id, attemptId: attempt.attemptId }, }); diff --git a/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts b/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts index e560e20..8f508c3 100644 --- a/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts +++ b/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts @@ -5,6 +5,7 @@ import { isDeepStrictEqual } from 'node:util'; import type { Prisma } from '../../../../../generated/prisma/client'; import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; +import { claimSyncWork, lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim'; import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; @@ -159,24 +160,18 @@ export class CatalogDiscoveryRepository { initiator: TbankSyncAttempt = attempt, ) { return this.prisma.$transaction(async (tx) => { - const origin = await tx.tbankSyncWork.updateMany({ - where: { - ...initiator, - state: { in: ['running', 'complete'] }, - connection: { invalidToken: false }, - }, - data: { progress: {} }, + const origin = await claimSyncWork( + tx, + initiator, + { progress: {} }, + { state: { in: ['running', 'complete'] } }, + ); + if (!origin) return false; + const claimed = await claimSyncWork(tx, attempt, { + state: 'complete', + progress: {}, }); - if (!origin.count) return false; - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', progress: {} }, - }); - if (!claimed.count) return false; + if (!claimed) return false; const payload = JSON.parse( JSON.stringify(value), ) as Prisma.InputJsonObject; @@ -284,11 +279,8 @@ export class CatalogDiscoveryRepository { active: ReadonlySet, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { ...owner, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!current.count) return null; + const current = await lockConnectionGeneration(tx, owner); + if (!current) return null; const scope = { ...owner, scopeKey: catalogScope(input) }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: scope }, @@ -341,15 +333,11 @@ export class CatalogDiscoveryRepository { now: Date, ) { return this.prisma.$transaction(async (tx) => { - const claimed = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { state: 'complete', progress: {} }, + const claimed = await claimSyncWork(tx, attempt, { + state: 'complete', + progress: {}, }); - if (!claimed.count) return false; + if (!claimed) return false; for (const instrument of instruments) { // Discovery can create a minimal identity, but cannot overwrite verified metadata. const identity = { provider, uid: instrument.uid }; @@ -461,6 +449,7 @@ export class CatalogDiscoveryRepository { finish(attempt: TbankSyncAttempt, success: boolean) { return this.prisma.$transaction(async (tx) => { + // Fails sibling attempts whose connection was invalidated mid-flight. await tx.tbankSyncWork.updateMany({ where: { ...attempt, @@ -469,16 +458,9 @@ export class CatalogDiscoveryRepository { }, data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken }, }); - await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { - state: success ? 'complete' : 'failed', - errorCode: success ? null : TbankErrorCode.Unavailable, - }, + await claimSyncWork(tx, attempt, { + state: success ? 'complete' : 'failed', + errorCode: success ? null : TbankErrorCode.Unavailable, }); }); } diff --git a/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts b/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts index 844d1a8..161661f 100644 --- a/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts +++ b/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts @@ -7,13 +7,14 @@ import { createHash, randomUUID } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import type { Prisma } from '../../../../../generated/prisma/client'; -import type { - TbankOperationItem, - TbankOperationsByCursorResponse, -} from '../../../../integrations/tbank/tbank.api'; import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; import { isTbankProtocolValue, TbankOperationState } from '../../../../integrations/tbank/tbank-protocol'; +import { claimSyncWork, lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim'; +import { + isJoinableRunningAttempt, + type TbankOperationItem, type TbankOperationsByCursorResponse, +} from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; import { dateOnlyBounds, @@ -268,16 +269,10 @@ export class BrokerageOperationsRepository { .update(JSON.stringify([...new Set(intervals.map(intervalKey))].sort())) .digest('hex'); return this.prisma.$transaction(async (tx) => { - const locked = await tx.tbankConnection.updateMany({ - where: { - userId, - revision, - invalidToken: false, - membershipAt: { not: null }, - }, - data: { membershipRevision: { increment: 0 } }, + const locked = await lockConnectionGeneration(tx, { userId, revision }, { + membershipAt: { not: null }, }); - if (!locked.count) return null; + if (!locked) return null; const accounts = await tx.tbankAccount.findMany({ where: { userId, revision }, select: { accountId: true, openedOn: true }, @@ -379,9 +374,7 @@ export class BrokerageOperationsRepository { publicationAtomic && previousIntent && !previousIntent.atomic; if ( touching.length <= SINGLE_WORK_GROUP && - previous?.state === 'running' && - previous.processRunId === processRunId && - runningAttempts.has(previous.attemptId) && + isJoinableRunningAttempt(previous, processRunId, runningAttempts) && !upgradeGroup ) { const pending = previousIntent?.intervals.map(decodeInterval) ?? []; @@ -933,15 +926,8 @@ async function claim( tx: Prisma.TransactionClient, attempt: OperationAttempt, ): Promise { - const result = await tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { updatedAt: new Date() }, - }); - return result.count === EXPECTED_UPDATED_WORK_ITEMS; + const count = await claimSyncWork(tx, attempt, { updatedAt: new Date() }); + return count === EXPECTED_UPDATED_WORK_ITEMS; } function decodeInterval(key: string): OperationInterval { diff --git a/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts b/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts index 3be02a0..1955ba5 100644 --- a/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts +++ b/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts @@ -5,7 +5,8 @@ import type { Prisma } from '../../../../../generated/prisma/client'; import type { PortfolioDto } from '../presentation/portfolio.dto'; import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; -import { tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api'; +import { lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim'; +import { isJoinableRunningAttempt, tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; const EXPECTED_UPDATED_ACCOUNTS = 1; @@ -56,16 +57,10 @@ export class PortfolioRepository { force: boolean, ) { return this.prisma.$transaction(async (tx) => { - const current = await tx.tbankConnection.updateMany({ - where: { - userId, - revision, - invalidToken: false, - accounts: { some: { accountId } }, - }, - data: { membershipRevision: { increment: 0 } }, + const current = await lockConnectionGeneration(tx, { userId, revision }, { + accounts: { some: { accountId } }, }); - if (!current.count) return null; + if (!current) return null; const snapshot = await tx.tbankPortfolio.findUnique({ where: { userId_revision_accountId: { userId, revision, accountId } }, }); @@ -86,11 +81,7 @@ export class PortfolioRepository { return null; } const scope = portfolioWorkScope; - if ( - previous?.state === 'running' && - previous.processRunId === processRunId && - pending.has(previous.attemptId) - ) { + if (isJoinableRunningAttempt(previous, processRunId, pending)) { return { work: previous, launch: false }; } const data = { diff --git a/src/modules/investments/tbank-sync/tbank-sync.repository.ts b/src/modules/investments/tbank-sync/tbank-sync.repository.ts index 8ccb815..fb1a872 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.repository.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.repository.ts @@ -6,6 +6,7 @@ import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api'; import type { InvestmentAttemptDto } from '../investment-resource.dto'; import type { InvestmentScreenDto } from './tbank-sync.dto'; +import { claimSyncWork, lockConnectionGeneration } from '../../../integrations/tbank/tbank-sync-claim'; import { PrismaService } from '../../../prisma/prisma.service'; export type ScreenIntent = { now: string; scope: InvestmentScreenDto }; @@ -88,11 +89,8 @@ export class TbankSyncRepository { runId: string, ) { return this.prisma.$transaction(async (tx) => { - const valid = await tx.tbankConnection.updateMany({ - where: { userId, revision, invalidToken: false }, - data: { membershipRevision: { increment: 0 } }, - }); - if (!valid.count) return null; + const valid = await lockConnectionGeneration(tx, { userId, revision }); + if (!valid) return null; const scopeKey = `screen:${JSON.stringify(scope)}`; const key = { userId, revision, scopeKey }; const previous = await tx.tbankSyncWork.findUnique({ @@ -176,17 +174,11 @@ export class TbankSyncRepository { ...Object.fromEntries(rows.map((row) => [row.id, row.attemptId])), ...old.attempts, }; - return tx.tbankSyncWork.updateMany({ - where: { - ...attempt, - state: 'running', - connection: { invalidToken: false }, - }, - data: { - progress: { children: [...new Set(children)], attempts, dispatched }, - ...(errorCode ? { state: 'failed', errorCode } : {}), - }, + const claimed = await claimSyncWork(tx, attempt, { + progress: { children: [...new Set(children)], attempts, dispatched }, + ...(errorCode ? { state: 'failed', errorCode } : {}), }); + return { count: claimed }; }); } @@ -364,20 +356,17 @@ export class TbankSyncRepository { where: { userId: attempt.userId, id: child.id }, select: { attemptId: true }, }); - const updated = await tx.tbankSyncWork.updateMany({ - where: { ...attempt, state: 'running' }, - data: { - progress: { - attempts: { - ...progress.attempts, - ...(row ? { [child.id]: row.attemptId } : {}), - }, - children: [...(progress.children ?? []), child], - sections: progress.sections ?? [], + const updated = await claimSyncWork(tx, attempt, { + progress: { + attempts: { + ...progress.attempts, + ...(row ? { [child.id]: row.attemptId } : {}), }, + children: [...(progress.children ?? []), child], + sections: progress.sections ?? [], }, - }); - return Boolean(updated.count); + }, { withoutConnectionFence: true }); + return Boolean(updated); }); }