From 1bc254ae80d9c3ece5a6b247894580ed75266b28 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Tue, 3 Nov 2026 22:32:35 +0300 Subject: [PATCH] refactor: share sync reservation join rule (WealthLedger/backend#119) --- .../tbank/instrument-metadata.repository.ts | 8 +- .../tbank/tbank-access.service.ts | 22 +++- src/integrations/tbank/tbank.api.ts | 1 + .../catalog-discovery.repository.ts | 8 +- test/tbank-membership-reservation.e2e-spec.ts | 107 ++++++++++++++++++ 5 files changed, 129 insertions(+), 17 deletions(-) create mode 100644 test/tbank-membership-reservation.e2e-spec.ts diff --git a/src/integrations/tbank/instrument-metadata.repository.ts b/src/integrations/tbank/instrument-metadata.repository.ts index 93577d9..73a0396 100644 --- a/src/integrations/tbank/instrument-metadata.repository.ts +++ b/src/integrations/tbank/instrument-metadata.repository.ts @@ -7,10 +7,10 @@ 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'; const FIRST_BATCH_OFFSET = 0; @@ -131,11 +131,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 = { diff --git a/src/integrations/tbank/tbank-access.service.ts b/src/integrations/tbank/tbank-access.service.ts index 06e22d2..98f030a 100644 --- a/src/integrations/tbank/tbank-access.service.ts +++ b/src/integrations/tbank/tbank-access.service.ts @@ -25,6 +25,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( @@ -62,11 +78,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 = { diff --git a/src/integrations/tbank/tbank.api.ts b/src/integrations/tbank/tbank.api.ts index 1f297d5..ecf0e3c 100644 --- a/src/integrations/tbank/tbank.api.ts +++ b/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'; diff --git a/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts b/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts index 23cbc0e..e560e20 100644 --- a/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts +++ b/src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts @@ -3,9 +3,9 @@ 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 { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; const UNCHANGED_REVISION_INCREMENT = 0; @@ -293,11 +293,7 @@ export class CatalogDiscoveryRepository { 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 = { diff --git a/test/tbank-membership-reservation.e2e-spec.ts b/test/tbank-membership-reservation.e2e-spec.ts new file mode 100644 index 0000000..3095691 --- /dev/null +++ b/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); + }); +}); -- 2.54.0