WealthLedger/backend
5
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: share tbank-sync claim and generation lock (WealthLedger/backend#166)

Browse Source
This commit is contained in:
ksv741
2026-11-04 03:23:41 +03:00
parent fa0a6e5603
commit 15fa6c6739
8 changed files with 171 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/investments/instruments/infrastructure/candle.repository.ts
+31 -51
View File
@@ -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
+22 -40
View File
@@ -5,6 +5,7 @@ import { isDeepStrictEqual } from 'node:util';
import type { Prisma } from '../../../../../generated/prisma/client';
import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
import { claimSyncWork, lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim';
import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
import { PrismaService } from '../../../../prisma/prisma.service';
@@ -159,24 +160,18 @@ export class CatalogDiscoveryRepository {
initiator: TbankSyncAttempt = attempt,
) {
return this.prisma.$transaction(async (tx) => {
const origin = await tx.tbankSyncWork.updateMany({
where: {
...initiator,
state: { in: ['running', 'complete'] },
connection: { invalidToken: false },
},
data: { progress: {} },
const origin = await claimSyncWork(
tx,
initiator,
{ progress: {} },
{ state: { in: ['running', 'complete'] } },
);
if (!origin) return false;
const claimed = await claimSyncWork(tx, attempt, {
state: 'complete',
progress: {},
});
if (!origin.count) return false;
const claimed = await tx.tbankSyncWork.updateMany({
where: {
...attempt,
state: 'running',
connection: { invalidToken: false },
},
data: { state: 'complete', progress: {} },
});
if (!claimed.count) return false;
if (!claimed) return false;
const payload = JSON.parse(
JSON.stringify(value),
) as Prisma.InputJsonObject;
@@ -284,11 +279,8 @@ export class CatalogDiscoveryRepository {
active: ReadonlySet<string>,
) {
return this.prisma.$transaction(async (tx) => {
const current = await tx.tbankConnection.updateMany({
where: { ...owner, invalidToken: false },
data: { membershipRevision: { increment: 0 } },
});
if (!current.count) return null;
const current = await lockConnectionGeneration(tx, owner);
if (!current) return null;
const scope = { ...owner, scopeKey: catalogScope(input) };
const previous = await tx.tbankSyncWork.findUnique({
where: { userId_revision_scopeKey: scope },
@@ -341,15 +333,11 @@ export class CatalogDiscoveryRepository {
now: Date,
) {
return this.prisma.$transaction(async (tx) => {
const claimed = await tx.tbankSyncWork.updateMany({
where: {
...attempt,
state: 'running',
connection: { invalidToken: false },
},
data: { state: 'complete', progress: {} },
const claimed = await claimSyncWork(tx, attempt, {
state: 'complete',
progress: {},
});
if (!claimed.count) return false;
if (!claimed) return false;
for (const instrument of instruments) {
// Discovery can create a minimal identity, but cannot overwrite verified metadata.
const identity = { provider, uid: instrument.uid };
@@ -461,6 +449,7 @@ export class CatalogDiscoveryRepository {
finish(attempt: TbankSyncAttempt, success: boolean) {
return this.prisma.$transaction(async (tx) => {
// Fails sibling attempts whose connection was invalidated mid-flight.
await tx.tbankSyncWork.updateMany({
where: {
...attempt,
@@ -469,16 +458,9 @@ export class CatalogDiscoveryRepository {
},
data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken },
});
await tx.tbankSyncWork.updateMany({
where: {
...attempt,
state: 'running',
connection: { invalidToken: false },
},
data: {
state: success ? 'complete' : 'failed',
errorCode: success ? null : TbankErrorCode.Unavailable,
},
await claimSyncWork(tx, attempt, {
state: success ? 'complete' : 'failed',
errorCode: success ? null : TbankErrorCode.Unavailable,
});
});
}
src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts
+11 -25
View File
@@ -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/infrastructure/portfolio.repository.ts
+6 -15
View File
@@ -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 = {
src/modules/investments/tbank-sync/tbank-sync.repository.ts
+16 -27
View File
@@ -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);
});
}