import { Injectable } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; 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 { PrismaService } from '../../../../prisma/prisma.service'; const EXPECTED_UPDATED_ACCOUNTS = 1; type PortfolioAttempt = { attemptId: string; id: string; revision: string; userId: string; }; export const portfolioFreshForMs = tbankSourceTtlMs.portfolio; @Injectable() export class PortfolioRepository { constructor(private readonly prisma: PrismaService) {} read(userId: string, revision: string) { return this.prisma.$transaction((tx) => tx.tbankConnection.findUnique({ where: { userId_revision: { userId, revision }, invalidToken: false }, select: { membershipAt: true, membershipRevision: true, accounts: { orderBy: { upstreamOrder: 'asc' }, include: { portfolio: true }, }, work: { where: { OR: [ { scopeKey: 'membership' }, { scopeKey: { startsWith: 'portfolio:' } }, ], }, select: { scopeKey: true, id: true, state: true, errorCode: true }, }, }, }), ); } reserve( userId: string, revision: string, accountId: string, processRunId: string, pending: ReadonlyMap>, 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 } }, }); if (!current.count) return null; const snapshot = await tx.tbankPortfolio.findUnique({ where: { userId_revision_accountId: { userId, revision, accountId } }, }); const portfolioWorkScope = { userId, revision, scopeKey: portfolioWorkKey(accountId), }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: portfolioWorkScope }, }); if ( !force && snapshot && !tbankSourceNeedsRefresh('portfolio', snapshot.fetchedAt, new Date()) && previous?.state !== 'failed' ) { return null; } const scope = portfolioWorkScope; if ( previous?.state === 'running' && previous.processRunId === processRunId && pending.has(previous.attemptId) ) { return { work: previous, launch: false }; } const data = { state: 'running', attemptId: randomUUID(), processRunId, priority: 1, intent: { source: 'portfolio', accountId }, progress: {}, errorCode: null, }; return { work: await tx.tbankSyncWork.upsert({ where: { userId_revision_scopeKey: scope }, create: { ...scope, ...data, id: randomUUID() }, update: data, }), launch: true, }; }); } publish( attempt: PortfolioAttempt, accountId: string, snapshot: PortfolioDto, ) { return this.prisma.$transaction(async (tx) => { const claimed = await tx.tbankSyncWork.updateMany({ where: { ...attempt, state: 'running', connection: { invalidToken: false, accounts: { some: { accountId } }, }, }, data: { state: 'complete', progress: {} }, }); if (!claimed.count) return false; const scope = { userId: attempt.userId, revision: attempt.revision, accountId, }; const data = { payload: JSON.parse(JSON.stringify(snapshot)) as Prisma.InputJsonObject, fetchedAt: new Date(snapshot.lastSuccessfulAt), }; await tx.tbankPortfolio.upsert({ where: { userId_revision_accountId: scope }, create: { ...scope, ...data }, update: { ...data, dataRevision: { increment: 1 } }, }); // Value history points require a fully valued snapshot; the attempt // identity deduplicates repeated or late publications of one attempt. if ( snapshot.totalsByCurrency.length > Number.MIN_VALUE && snapshot.positions.every((position) => position.currentValue !== null) ) { await tx.tbankPortfolioValueHistory.upsert({ where: { userId_accountId: { userId: attempt.userId, accountId }, }, create: { userId: attempt.userId, accountId }, update: { deletedAt: null }, }); for (const { currency, value } of snapshot.totalsByCurrency) { await tx.tbankPortfolioValuePoint.upsert({ where: { userId_accountId_attemptId_currency: { userId: attempt.userId, accountId, attemptId: attempt.attemptId, currency, }, }, create: { userId: attempt.userId, accountId, attemptId: attempt.attemptId, currency, value, capturedAt: data.fetchedAt, }, update: {}, }); } } return true; }); } async fail(attempt: PortfolioAttempt) { const result = await this.prisma.tbankSyncWork.updateMany({ where: { ...attempt, state: 'running', connection: { invalidToken: false }, }, data: { state: 'failed', errorCode: TbankErrorCode.Unavailable }, }); return result.count === EXPECTED_UPDATED_ACCOUNTS; } } export function portfolioWorkKey(accountId: string): string { return `portfolio:${JSON.stringify(accountId)}`; }