import { Injectable } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; import type { Prisma } from '../../../../generated/prisma/client'; import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api'; import type { InvestmentAttemptDto } from '../investment-resource.dto'; import type { InvestmentScreenDto } from './tbank-sync.dto'; import { PrismaService } from '../../../prisma/prisma.service'; export type ScreenIntent = { now: string; scope: InvestmentScreenDto }; export type ScreenProgress = { attempts?: Record; children: string[]; dispatched: boolean; }; export type HourlySection = { accountId?: string; errorCode: null | string; section: 'events' | 'membership' | 'operations' | 'portfolio'; }; export type HourlyProgress = { result?: 'failed' | 'partial' | 'succeeded'; sections: HourlySection[]; }; type HourlyAttempt = { finishedAt: Date | null; sections: HourlySection[]; startedAt: Date; state: 'cancelled' | 'failed' | 'partial' | 'running' | 'succeeded'; }; export function hourlyAttempt( hourly: HourlyAttempt | null, section: HourlySection['section'], accountId?: string, ): InvestmentAttemptDto { if (!hourly) { return { state: 'none', startedAt: null, finishedAt: null, problems: [] }; } const problems = hourly.sections .filter( (entry) => entry.section === section && (accountId === undefined || entry.accountId === accountId), ) .map((entry) => ({ section: entry.section, accountId: entry.accountId ?? null, errorCode: entry.errorCode, })); return { state: hourly.state === 'cancelled' ? 'failed' : hourly.state, startedAt: hourly.startedAt.toISOString(), finishedAt: hourly.finishedAt?.toISOString() ?? null, problems, }; } @Injectable() export class TbankSyncRepository { constructor(private readonly prisma: PrismaService) {} reserve( userId: string, revision: string, scope: InvestmentScreenDto, 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 scopeKey = `screen:${JSON.stringify(scope)}`; const key = { userId, revision, scopeKey }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: key }, }); if (previous?.state === 'running' && previous.processRunId === runId) { const progress = previous.progress as ScreenProgress; const children = progress.children ?? []; if ( !progress.dispatched || (await tx.tbankSyncWork.count({ where: { userId, revision, id: { in: children }, state: 'running' }, })) ) { return { work: previous, launch: false }; } } const data = { attemptId: randomUUID(), processRunId: runId, state: 'running', priority: 0, intent: JSON.parse( JSON.stringify({ scope, now: new Date().toISOString() }), ) as Prisma.InputJsonObject, progress: { children: [], dispatched: false }, errorCode: null, }; const work = await tx.tbankSyncWork.upsert({ where: { userId_revision_scopeKey: key }, create: { ...key, ...data, id: randomUUID() }, update: { ...data, id: randomUUID() }, }); return { work, launch: true }; }); } read(userId: string, revision: string, id: string) { return this.prisma.$transaction(async (tx) => { const work = await tx.tbankSyncWork.findFirst({ where: { userId, revision, id }, }); if (!work) return null; const progress = work.progress as ScreenProgress; const children = await tx.tbankSyncWork.findMany({ where: { userId, revision, id: { in: progress.children ?? [] } }, select: { id: true, attemptId: true, state: true, errorCode: true }, }); return { work, children: children.map((child) => progress.attempts?.[child.id] && progress.attempts[child.id] !== child.attemptId ? { ...child, state: 'cancelled', errorCode: null } : child, ), progress, }; }); } async progress( attempt: TbankSyncAttempt, children: string[], dispatched: boolean, errorCode: null | string = null, ) { return this.prisma.$transaction(async (tx) => { const previous = await tx.tbankSyncWork.findFirst({ where: attempt }); if (!previous) return { count: 0 }; const old = previous.progress as ScreenProgress; const rows = await tx.tbankSyncWork.findMany({ where: { userId: attempt.userId, revision: attempt.revision, id: { in: children }, }, select: { id: true, attemptId: true }, }); const attempts = { ...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 } : {}), }, }); }); } // The timer only proposes work; this reservation is the durable fence. The // attempt start (createdAt) is the basis of the next hourly term, including // failed and interrupted attempts. reserveHourly(userId: string, runId: string, hourMs: number) { return this.prisma.$transaction(async (tx) => { const connection = await tx.tbankConnection.findFirst({ where: { userId, invalidToken: false }, select: { revision: true }, }); if (!connection) return null; const key = { userId, revision: connection.revision, scopeKey: 'hourly', }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: key }, }); const now = new Date(); if (previous) { const due = now.valueOf() - previous.createdAt.valueOf() >= hourMs; if (previous.state === 'running') { if (previous.processRunId === runId) return null; // A dead process cannot finish its attempt; interrupt it, but its // start stays the basis of the next term. await tx.tbankSyncWork.update({ where: { userId_revision_scopeKey: key }, data: { state: 'cancelled', errorCode: null }, }); if (!due) return null; } else if (!due) return null; } const data = { attemptId: randomUUID(), processRunId: runId, state: 'running', priority: 0, intent: JSON.parse( JSON.stringify({ source: 'hourly' }), ) as Prisma.InputJsonObject, progress: { sections: [] } as unknown as Prisma.InputJsonValue, errorCode: null, }; const work = await tx.tbankSyncWork.upsert({ where: { userId_revision_scopeKey: key }, create: { ...key, ...data, id: randomUUID() }, update: { ...data, createdAt: now }, }); return { work, launch: true }; }); } settleHourly( attempt: TbankSyncAttempt, result: 'failed' | 'partial' | 'succeeded', sections: HourlySection[], ) { return this.prisma.tbankSyncWork.updateMany({ where: { id: attempt.id, attemptId: attempt.attemptId, userId: attempt.userId, revision: attempt.revision, state: 'running', }, data: { state: result === 'failed' ? 'failed' : 'complete', errorCode: sections.find((section) => section.errorCode)?.errorCode ?? null, progress: { result, sections }, }, }); } async hourly(userId: string) { const connection = await this.prisma.tbankConnection.findFirst({ where: { userId }, select: { revision: true, invalidToken: true }, }); if (!connection || connection.invalidToken) return null; const work = await this.prisma.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: { userId, revision: connection.revision, scopeKey: 'hourly', }, }, }); if (!work) return null; const progress = work.progress as HourlyProgress; let state: 'cancelled' | 'running' | HourlyProgress['result']; if (work.state === 'running') state = 'running'; else if (work.state === 'cancelled') state = 'cancelled'; else state = progress.result ?? 'failed'; return { startedAt: work.createdAt, finishedAt: work.state === 'running' ? null : work.updatedAt, state, sections: progress.sections ?? [], }; } works(userId: string, ids: string[]) { return this.prisma.tbankSyncWork.findMany({ where: { userId, id: { in: ids } }, select: { id: true, state: true, errorCode: true }, }); } async workError(userId: string, id: string) { const work = await this.prisma.tbankSyncWork.findFirst({ where: { userId, id }, select: { state: true, errorCode: true }, }); return work?.state === 'failed' ? work.errorCode : null; } }