import { Injectable } from '@nestjs/common'; import { createHash, randomUUID } from 'node:crypto'; import type { Prisma } from '../../../../../generated/prisma/client'; import type { TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api'; import { PrismaService } from '../../../../prisma/prisma.service'; import { type CandleRange, confirmedCandleCoverage, type SavedCandle, } from '../domain/candle-coverage'; const NO_UPDATED_ROWS = 0; const provider = 'tbank'; const interval = 'day'; type CandleIntent = { force?: boolean; instrumentType?: 'bond' | 'share'; now?: string; period?: string; ranges: CandleRange[]; uid: string; }; @Injectable() export class CandleRepository { constructor(private readonly prisma: PrismaService) {} read(uid: string, range: CandleRange) { const identity = { provider, instrumentUid: uid, interval }; return this.prisma.$transaction(async (tx) => { const candles = await tx.tbankCandle.findMany({ where: { ...identity, time: { gte: range.from, lt: range.to } }, orderBy: { time: 'asc' }, }); const coverage = await tx.tbankCandleCoverage.findMany({ where: { ...identity, from: { lt: range.to }, to: { gt: range.from } }, orderBy: [{ from: 'asc' }, { to: 'asc' }], }); const data = candles.map((row) => ({ time: row.time.toISOString(), close: row.close, isComplete: row.isComplete, })); return { data, coverage, dataRevision: createHash('sha256') .update(JSON.stringify(data)) .digest('base64url'), }; }); } reserve( owner: { revision: string; userId: string }, scopeKey: string, intent: CandleIntent, runId: string, active: ReadonlySet, priority = 1, ) { return this.prisma.$transaction(async (tx) => { const current = await tx.tbankConnection.updateMany({ where: { ...owner, invalidToken: false }, data: { membershipRevision: { increment: 0 } }, }); if (!current.count) return null; const scope = { ...owner, scopeKey }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: scope }, }); if ( previous?.state === 'running' && previous.processRunId === runId && active.has(previous.attemptId) ) { const previousIntent = previous.intent as Prisma.JsonObject; const previousNow = typeof previousIntent.now === 'string' ? previousIntent.now : ''; const nextPriority = Math.min(previous.priority, priority); if ( nextPriority === previous.priority && (!intent.force || (previousIntent.force === true && (!intent.now || intent.now <= previousNow))) ) { return { work: previous, launch: false }; } const work = await tx.tbankSyncWork.update({ where: { id: previous.id }, data: { priority: nextPriority, intent: { ...previousIntent, ...(intent.force ? { force: true, ...(intent.now && intent.now > previousNow ? { now: intent.now } : {}), } : {}), }, }, }); return { work, launch: false }; } if (previous) { await tx.tbankSyncStage.deleteMany({ where: { workId: previous.id } }); } const data = { attemptId: randomUUID(), processRunId: runId, state: 'running', priority, intent: JSON.parse(JSON.stringify(intent)) as Prisma.InputJsonObject, progress: {}, errorCode: null, }; const work = await tx.tbankSyncWork.upsert({ where: { userId_revision_scopeKey: scope }, create: { ...scope, ...data, id: randomUUID() }, update: data, }); return { work, launch: true }; }); } work(userId: string, revision: string, scopeKey: string) { return this.prisma.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: { userId, revision, scopeKey } }, }); } stage( attempt: TbankSyncAttempt, range: CandleRange, candles: SavedCandle[], fetchedAt: Date, ) { // Reject malformed financial input instead of publishing a false empty range. for (const candle of candles) { if ( !Number.isFinite(Date.parse(candle.time)) || // eslint-disable-next-line security/detect-unsafe-regex -- Anchored decimal grammar: the literal decimal point separates disjoint digit runs; backtracking is linear. !/^-?\d+(?:\.\d+)?$/.test(candle.close) ) { throw new Error('Invalid candle'); } } const key = `${range.from.toISOString()}/${range.to.toISOString()}`; const rows = [ ...new Map( candles .filter( (candle) => new Date(candle.time) >= range.from && new Date(candle.time) < range.to, ) .map((candle) => [candle.time, candle]), ).values(), ]; return this.prisma.$transaction(async (tx) => { const valid = await tx.tbankSyncWork.updateMany({ where: { ...attempt, state: 'running', connection: { invalidToken: false }, }, data: { progress: { lastWindow: key } }, }); if (!valid.count) return false; const identity = { workId: attempt.id, attemptId: attempt.attemptId, sourceKey: 'candles', itemKey: key, }; const payload = JSON.parse( JSON.stringify({ range, candles: rows, fetchedAt }), ) as Prisma.InputJsonObject; await tx.tbankSyncStage.upsert({ where: { workId_attemptId_sourceKey_itemKey: identity }, create: { ...identity, payload }, update: { payload }, }); return true; }); } publish( attempt: TbankSyncAttempt, intent: CandleIntent, completeWork = true, ) { return this.prisma.$transaction(async (tx) => { const claimed = await tx.tbankSyncWork.updateMany({ where: { ...attempt, state: 'running', connection: { invalidToken: false }, }, data: { ...(completeWork ? { state: 'complete' } : {}), errorCode: null, }, }); if (!claimed.count) return false; const staged = await tx.tbankSyncStage.findMany({ where: { workId: attempt.id, attemptId: attempt.attemptId, sourceKey: 'candles', }, }); if ( intent.ranges.some( (range) => !staged.some( (row) => row.itemKey === `${range.from.toISOString()}/${range.to.toISOString()}`, ), ) ) { throw new Error('Incomplete candle reconciliation'); } const identity = { provider, instrumentUid: intent.uid, interval }; for (const range of intent.ranges) { const row = staged.find( (entry) => entry.itemKey === `${range.from.toISOString()}/${range.to.toISOString()}`, ); if (!row) throw new Error('Incomplete candle reconciliation'); const payload = row.payload as { candles: SavedCandle[]; fetchedAt: string; }; const fetchedAt = new Date(payload.fetchedAt); const old = await tx.tbankCandleCoverage.findMany({ where: { ...identity, from: { lt: range.to }, to: { gt: range.from }, }, }); await tx.tbankCandleCoverage.deleteMany({ where: { ...identity, from: { lt: range.to }, to: { gt: range.from }, }, }); // Preserve coverage outside the requested reconciliation boundaries. for (const previous of old) { for (const part of [ ...(previous.from < range.from ? [{ from: previous.from, to: range.from }] : []), ...(previous.to > range.to ? [{ from: range.to, to: previous.to }] : []), ]) { await tx.tbankCandleCoverage.create({ data: { ...previous, ...part }, }); } } await tx.tbankCandle.deleteMany({ where: { ...identity, time: { gte: range.from, lt: range.to } }, }); for (const candle of payload.candles) { await tx.tbankCandle.create({ data: { ...identity, time: new Date(candle.time), close: candle.close, isComplete: candle.isComplete, payload: candle, fetchedAt, }, }); } for (const covered of confirmedCandleCoverage( range, payload.candles, fetchedAt, )) { await tx.tbankCandleCoverage.create({ data: { ...identity, ...covered }, }); } } await tx.tbankSyncStage.deleteMany({ where: { workId: attempt.id, attemptId: attempt.attemptId }, }); return true; }); } completeObserver(attempt: TbankSyncAttempt, initiator: TbankSyncAttempt) { return this.prisma.$transaction(async (tx) => { const origin = await tx.tbankSyncWork.findFirst({ where: { ...initiator, state: 'complete', connection: { invalidToken: false }, }, }); if (!origin) return false; const result = await tx.tbankSyncWork.updateMany({ where: { ...attempt, state: 'running', connection: { invalidToken: false }, }, data: { state: 'complete', errorCode: null }, }); return result.count > NO_UPDATED_ROWS; }); } fail(attempt: TbankSyncAttempt, errorCode: string) { return this.prisma.$transaction(async (tx) => { await tx.tbankSyncWork.updateMany({ where: { ...attempt, state: 'running', connection: { invalidToken: false }, }, data: { state: 'failed', errorCode }, }); await tx.tbankSyncStage.deleteMany({ where: { workId: attempt.id, attemptId: attempt.attemptId }, }); }); } }