import { Injectable } from '@nestjs/common'; import { createHash, randomUUID } from 'node:crypto'; import type { Prisma } from '#generated/prisma/client'; import { claimSyncWork, lockConnectionGeneration, } from '#src/integrations/tbank/tbank-sync-claim'; import { isJoinableRunningAttempt, type TbankSyncAttempt, } from '#src/integrations/tbank/tbank.api'; import { PrismaService } from '#src/prisma/prisma.service'; import { toJson } from '#src/prisma/to-json'; 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 lockConnectionGeneration(tx, owner); if (!current) return null; const scope = { ...owner, scopeKey }; const previous = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: scope }, }); const joinable = isJoinableRunningAttempt(previous, runId, active) ? previous : null; if (joinable) { const previousIntent = joinable.intent as Prisma.JsonObject; const previousNow = typeof previousIntent.now === 'string' ? previousIntent.now : ''; const nextPriority = Math.min(joinable.priority, priority); if ( nextPriority === joinable.priority && (!intent.force || (previousIntent.force === true && (!intent.now || intent.now <= previousNow))) ) { return { work: joinable, launch: false }; } const work = await tx.tbankSyncWork.update({ where: { id: joinable.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: toJson(intent), 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 claimSyncWork(tx, attempt, { progress: { lastWindow: key }, }); if (!valid) return false; const identity = { workId: attempt.id, attemptId: attempt.attemptId, sourceKey: 'candles', itemKey: key, }; const payload = toJson({ range, candles: rows, fetchedAt }); 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 claimSyncWork(tx, attempt, { ...(completeWork ? { state: 'complete' } : {}), errorCode: null, }); if (!claimed) 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 claimSyncWork(tx, attempt, { state: 'complete', errorCode: null, }); return result > NO_UPDATED_ROWS; }); } fail(attempt: TbankSyncAttempt, errorCode: string) { return this.prisma.$transaction(async (tx) => { await claimSyncWork(tx, attempt, { state: 'failed', errorCode }); await tx.tbankSyncStage.deleteMany({ where: { workId: attempt.id, attemptId: attempt.attemptId }, }); }); } }