import { BadRequestException, ConflictException, Injectable, NotFoundException, } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; import type { TbankAccount } from './tbank.client'; 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; const MEMBERSHIP_REVISION_INCREMENT = 1; const UNCHANGED_MEMBERSHIP_REVISION_INCREMENT = 0; export type TbankSyncAttempt = { attemptId: string; id: string; revision: string; 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( private readonly prisma: PrismaService, private readonly tokens: TbankTokenService, ) {} async reserveMembership( userId: string, revision: string, processRunId: string, active: ReadonlySet, force: boolean, ) { return this.prisma.$transaction(async (tx) => { const current = await lockConnectionGeneration(tx, { userId, revision }); if (!current) return null; const connection = await tx.tbankConnection.findUniqueOrThrow({ where: { userId }, }); if ( !force && !tbankSourceNeedsRefresh( 'membership', connection.membershipAt, new Date(), ) ) { return null; } const scope = { userId, revision, scopeKey: 'membership' }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: scope }, }); if (isJoinableRunningAttempt(previous, processRunId, active)) { return { work: previous, launch: false }; } const data = { attemptId: randomUUID(), processRunId, state: 'running', intent: { source: 'membership' }, progress: {}, priority: 0, errorCode: null, }; const work = await tx.tbankSyncWork.upsert({ where: { userId_revision_scopeKey: scope }, create: { ...scope, ...data, id: randomUUID() }, update: data, }); return { work, launch: true }; }); } membershipWork(userId: string, revision: string) { return this.prisma.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: { userId, revision, scopeKey: 'membership' }, }, select: { id: true, state: true, errorCode: true, updatedAt: true }, }); } async failMembership(attempt: TbankSyncAttempt, invalidToken: boolean) { if (invalidToken) return this.rejectMembershipToken(attempt); const result = await claimSyncWork(this.prisma, attempt, { state: 'failed', errorCode: TbankErrorCode.Unavailable, }); return result === EXPECTED_UPDATED_CONNECTIONS; } async tokenForAttempt(attempt: TbankSyncAttempt): Promise { const work = await this.prisma.tbankSyncWork.findFirst({ where: { ...attempt, state: 'running', connection: { invalidToken: false }, }, select: { connection: { select: { encryptedToken: true } } }, }); return work ? this.tokens.decrypt(work.connection.encryptedToken) : null; } // Online reads: a permanent upstream auth refusal becomes 400/INVALID_TBANK_TOKEN here, once, for every caller. async online( userId: string, run: (connection: { revision: string; token: string }) => Promise, ): Promise { const connection = await this.prisma.tbankConnection.findUnique({ where: { userId }, select: { encryptedToken: true, invalidToken: true, revision: true }, }); if (!connection) throw new ConflictException({ code: TbankErrorCode.NotConnected }); if (connection.invalidToken) { throw new BadRequestException({ code: TbankErrorCode.InvalidToken }); } const token = this.tokens.decrypt(connection.encryptedToken); return this.tokens.translateInvalidToken(() => run({ revision: connection.revision, token })); } async publishMembership( attempt: TbankSyncAttempt, accounts: TbankAccount[], fetchedAt: Date, completeWork = false, ): Promise { const supported = supportedTbankAccounts(accounts); if (new Set(supported.map(({ id }) => id)).size !== supported.length) { throw new Error('Duplicate T-Bank account identity'); } 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 claimSyncWork(tx, attempt, { updatedAt: new Date() }); if (!claimed) return false; const { userId, revision } = attempt; const previous = await tx.tbankAccount.findMany({ where: { userId, revision }, orderBy: { upstreamOrder: 'asc' }, select: { accountId: true, type: true, name: true, openedOn: true }, }); const next = supported.map(({ id, ...account }) => ({ accountId: id, ...account, })); const changed = previous.length !== next.length || previous.some((account, index) => { // eslint-disable-next-line security/detect-object-injection -- The index comes from the paired account array iteration. const incoming = next[index]; return ( account.accountId !== incoming.accountId || account.type !== incoming.type || account.name !== incoming.name || account.openedOn !== incoming.openedOn ); }); await tx.tbankAccount.deleteMany({ where: { userId, revision, accountId: { notIn: supported.map(({ id }) => id) }, }, }); for (const [upstreamOrder, account] of next.entries()) { await tx.tbankAccount.upsert({ where: { userId_revision_accountId: { userId, revision, accountId: account.accountId, }, }, create: { userId, revision, ...account, upstreamOrder }, update: { ...account, upstreamOrder }, }); } await tx.tbankConnection.update({ where: { userId_revision: { userId, revision } }, data: { membershipAt: fetchedAt, membershipRevision: { increment: changed ? MEMBERSHIP_REVISION_INCREMENT : UNCHANGED_MEMBERSHIP_REVISION_INCREMENT }, }, }); // Value history visibility follows the confirmed membership only; a // failed fetch never reaches this transaction and changes nothing. const supportedIds = supported.map(({ id }) => id); await tx.tbankPortfolioValueHistory.updateMany({ where: { userId, accountId: { in: supportedIds } }, data: { deletedAt: null }, }); await tx.tbankPortfolioValueHistory.updateMany({ where: { userId, accountId: { notIn: supportedIds }, deletedAt: null }, data: { deletedAt: fetchedAt }, }); if (completeWork) { await tx.tbankSyncWork.update({ where: { id: attempt.id }, data: { state: 'complete', progress: { accounts: supported.length } }, }); } return true; }); } async rejectMembershipToken(attempt: TbankSyncAttempt): Promise { return this.prisma.$transaction(async (tx) => { const claimed = await claimSyncWork(tx, attempt, { state: 'failed', errorCode: TbankErrorCode.InvalidToken, }, { withoutConnectionFence: true }); if (!claimed) return false; await tx.tbankConnection.update({ where: { userId_revision: { userId: attempt.userId, revision: attempt.revision, }, }, data: { invalidToken: true }, }); return true; }); } // This boundary neither decrypts credentials nor starts discovery/sync. // Missing membership is distinct from a successfully observed empty list. async local(userId: string, accountId?: string) { const connection = await this.prisma.tbankConnection.findUnique({ where: { userId }, select: { userId: true, revision: true, invalidToken: true, membershipAt: true, membershipRevision: true, accounts: { orderBy: { upstreamOrder: 'asc' } }, }, }); if (!connection) { throw new ConflictException({ code: TbankErrorCode.NotConnected }); } if (connection.invalidToken) { throw new BadRequestException({ code: TbankErrorCode.InvalidToken }); } if ( connection.membershipAt && accountId && accountId !== 'all' && !connection.accounts.some((account) => account.accountId === accountId) ) { throw new NotFoundException(); } return connection; } }