backend/src/modules/investments/instruments/infrastructure/candle.repository.ts

import { Injectable } from '@nestjs/common';
import { createHash, randomUUID } from 'node:crypto';

import type { Prisma } from '../../../../../generated/prisma/client';

import {
  claimSyncWork,
  lockConnectionGeneration,
} from '../../../../integrations/tbank/tbank-sync-claim';
import {
  isJoinableRunningAttempt,
  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<string>,
    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: 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 claimSyncWork(tx, attempt, {
        progress: { lastWindow: key },
      });
      if (!valid) 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 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 },
      });
    });
  }
}