backend/src/modules/investments/portfolio/infrastructure/portfolio.repository.ts

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

import type { Prisma } from '../../../../../generated/prisma/client';
import type { PortfolioDto } from '../presentation/portfolio.dto';

import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
import { tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api';
import { PrismaService } from '../../../../prisma/prisma.service';

const EXPECTED_UPDATED_ACCOUNTS = 1;

type PortfolioAttempt = {
  attemptId: string;
  id: string;
  revision: string;
  userId: string;
};
export const portfolioFreshForMs = tbankSourceTtlMs.portfolio;

@Injectable()
export class PortfolioRepository {
  constructor(private readonly prisma: PrismaService) {}

  read(userId: string, revision: string) {
    return this.prisma.$transaction((tx) =>
      tx.tbankConnection.findUnique({
        where: { userId_revision: { userId, revision }, invalidToken: false },
        select: {
          membershipAt: true,
          membershipRevision: true,
          accounts: {
            orderBy: { upstreamOrder: 'asc' },
            include: { portfolio: true },
          },
          work: {
            where: {
              OR: [
                { scopeKey: 'membership' },
                { scopeKey: { startsWith: 'portfolio:' } },
              ],
            },
            select: { scopeKey: true, id: true, state: true, errorCode: true },
          },
        },
      }),
    );
  }

  reserve(
    userId: string,
    revision: string,
    accountId: string,
    processRunId: string,
    pending: ReadonlyMap<string, Promise<void>>,
    force: boolean,
  ) {
    return this.prisma.$transaction(async (tx) => {
      const current = await tx.tbankConnection.updateMany({
        where: {
          userId,
          revision,
          invalidToken: false,
          accounts: { some: { accountId } },
        },
        data: { membershipRevision: { increment: 0 } },
      });
      if (!current.count) return null;
      const snapshot = await tx.tbankPortfolio.findUnique({
        where: { userId_revision_accountId: { userId, revision, accountId } },
      });
      if (
        !force &&
        snapshot &&
        !tbankSourceNeedsRefresh('portfolio', snapshot.fetchedAt, new Date())
      ) {
        return null;
      }
      const scope = { userId, revision, scopeKey: portfolioWorkKey(accountId) };
      const previous = await tx.tbankSyncWork.findUnique({
        where: { userId_revision_scopeKey: scope },
      });
      if (
        previous?.state === 'running' &&
        previous.processRunId === processRunId &&
        pending.has(previous.attemptId)
      ) {
        return { work: previous, launch: false };
      }
      const data = {
        state: 'running',
        attemptId: randomUUID(),
        processRunId,
        priority: 1,
        intent: { source: 'portfolio', accountId },
        progress: {},
        errorCode: null,
      };
      return {
        work: await tx.tbankSyncWork.upsert({
          where: { userId_revision_scopeKey: scope },
          create: { ...scope, ...data, id: randomUUID() },
          update: data,
        }),
        launch: true,
      };
    });
  }

  publish(
    attempt: PortfolioAttempt,
    accountId: string,
    snapshot: PortfolioDto,
  ) {
    return this.prisma.$transaction(async (tx) => {
      const claimed = await tx.tbankSyncWork.updateMany({
        where: {
          ...attempt,
          state: 'running',
          connection: {
            invalidToken: false,
            accounts: { some: { accountId } },
          },
        },
        data: { state: 'complete', progress: {} },
      });
      if (!claimed.count) return false;
      const scope = {
        userId: attempt.userId,
        revision: attempt.revision,
        accountId,
      };
      const data = {
        payload: JSON.parse(JSON.stringify(snapshot)) as Prisma.InputJsonObject,
        fetchedAt: new Date(snapshot.lastSuccessfulAt),
      };
      await tx.tbankPortfolio.upsert({
        where: { userId_revision_accountId: scope },
        create: { ...scope, ...data },
        update: { ...data, dataRevision: { increment: 1 } },
      });
      return true;
    });
  }

  async fail(attempt: PortfolioAttempt) {
    const result = await this.prisma.tbankSyncWork.updateMany({
      where: {
        ...attempt,
        state: 'running',
        connection: { invalidToken: false },
      },
      data: { state: 'failed', errorCode: TbankErrorCode.Unavailable },
    });
    return result.count === EXPECTED_UPDATED_ACCOUNTS;
  }
}

export function portfolioWorkKey(accountId: string): string {
  return `portfolio:${JSON.stringify(accountId)}`;
}