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 }; // Json columns are written only by this repository; a shape check turns corruption into an explicit 500. export function screenIntent(value: Prisma.JsonValue): ScreenIntent { const intent = value as null | Partial; if (typeof intent?.now !== 'string' || typeof intent.scope?.screen !== 'string') { throw new Error('Corrupted screen sync intent'); } return intent as ScreenIntent; } export type ScreenProgress = { attempts?: Record; children: string[]; dispatched: boolean; }; export type HourlySection = { accountId?: string; errorCode: null | string; section: 'events' | 'membership' | 'operations' | 'portfolio'; }; export type HourlyChild = { accountId?: string; id: string; section: HourlySection['section']; }; export const HOURLY_CAUSE_DEADLINE = 'DEADLINE'; export const HOURLY_CAUSE_INTERRUPTED = 'INTERRUPTED'; export type HourlyCause = typeof HOURLY_CAUSE_INTERRUPTED; type StepOutcome = { problem?: HourlySection; success: boolean }; type StepRow = { attemptId: string; errorCode: null | string; state: string }; export type HourlyProgress = { attempts?: Record; children?: HourlyChild[]; 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'] | readonly HourlySection['section'][], accountId?: string, ): InvestmentAttemptDto { if (!hourly) { return { state: 'none', startedAt: null, finishedAt: null, problems: [] }; } const sections = Array.isArray(section) ? section : [section]; const problems = hourly.sections .filter( (entry) => sections.includes(entry.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. After downtime the next tick is due at // once, so no catch-up queue forms. 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) { // Terms are hourMs steps from the attempt start; a walk still running // at a term consumes it. A dead process's attempt has no known end, so // it consumes only its own term. const started = previous.createdAt.valueOf(); const ended = previous.state === 'running' ? started : previous.updatedAt.valueOf(); const dueAt = started + Math.max(hourMs, Math.ceil((ended - started) / hourMs) * hourMs); const due = now.valueOf() >= dueAt; if (previous.state === 'running') { if (previous.processRunId === runId) return null; // A dead process cannot finish its attempt; close it from its // actual step results, but its start stays the basis of the next // term. await this.closeHourly(tx, previous, 'INTERRUPTED', []); if (!due) return null; } else if (!due) return null; } const data = { attemptId: randomUUID(), processRunId: runId, state: 'running', priority: 0, intent: { source: 'hourly' }, progress: { sections: [] }, 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 }; }); } private classifyStep( child: HourlyChild, row: StepRow | undefined, recorded: string | undefined, cause: HourlyCause | null, ): StepOutcome { if (row && row.attemptId === recorded && row.state === 'complete') { return { success: true }; } const stale = !row || (recorded !== undefined && row.attemptId !== recorded); let errorCode: null | string = null; if (!stale && row.state === 'failed') errorCode = row.errorCode; else if (!stale && cause === HOURLY_CAUSE_INTERRUPTED) { errorCode = HOURLY_CAUSE_INTERRUPTED; } else if (!stale) errorCode = HOURLY_CAUSE_DEADLINE; return { problem: { accountId: child.accountId, errorCode, section: child.section }, success: false, }; } private deriveResult( cause: HourlyCause | null, sections: HourlySection[], successes: number, ): 'failed' | 'partial' | 'succeeded' { if (!cause && !sections.length) return 'succeeded'; return successes ? 'partial' : 'failed'; } // The outcome is derived from the durable state of the dispatched steps, // never from an in-memory error list that an exception path can empty. // `cause` marks a close before normal completion and caps the result at // `partial`; `dispatchProblems` records required sections that never got a // durable step row because their dispatch failed. private async closeHourly( tx: Prisma.TransactionClient, work: { attemptId: string; id: string; progress: Prisma.JsonValue; revision: string; userId: string; }, cause: HourlyCause | null, dispatchProblems: HourlySection[], ) { const progress = work.progress as HourlyProgress; const children = progress.children ?? []; const rows = children.length ? await tx.tbankSyncWork.findMany({ where: { userId: work.userId, id: { in: children.map((child) => child.id) }, }, select: { attemptId: true, errorCode: true, id: true, state: true }, }) : []; const byId = new Map(rows.map((row) => [row.id, row])); const outcomes = children.map((child) => this.classifyStep( child, byId.get(child.id), progress.attempts?.[child.id], cause, ), ); const sections = outcomes .flatMap((outcome) => (outcome.problem ? [outcome.problem] : [])) .concat(dispatchProblems); const successes = outcomes.filter((outcome) => outcome.success).length; const result = this.deriveResult(cause, sections, successes); return tx.tbankSyncWork.updateMany({ where: { id: work.id, attemptId: work.attemptId, state: 'running' }, data: { state: result === 'failed' ? 'failed' : 'complete', errorCode: cause ?? sections.find((section) => section.errorCode)?.errorCode ?? null, progress: { attempts: progress.attempts ?? {}, children, result, sections, }, }, }); } async settleHourly( attempt: TbankSyncAttempt, cause: HourlyCause | null = null, dispatchProblems: HourlySection[] = [], ) { return this.prisma.$transaction(async (tx) => { const work = await tx.tbankSyncWork.findFirst({ where: attempt }); if (!work || work.state !== 'running') return { count: 0 }; return this.closeHourly(tx, work, cause, dispatchProblems); }); } // Records a dispatched step on the attempt before it is awaited, so a lost // walk can still derive its outcome from durable rows. async recordHourlyChild(attempt: TbankSyncAttempt, child: HourlyChild) { return this.prisma.$transaction(async (tx) => { const work = await tx.tbankSyncWork.findFirst({ where: { ...attempt, state: 'running' }, select: { progress: true }, }); if (!work) return false; const progress = work.progress as HourlyProgress; const row = await tx.tbankSyncWork.findFirst({ 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 ?? [], }, }, }); return Boolean(updated.count); }); } runningHourly(userId: string) { return this.prisma.tbankSyncWork.findMany({ where: { userId, scopeKey: 'hourly', state: 'running' }, select: { attemptId: true, id: true, processRunId: true, revision: true }, }); } 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; } }