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

import { Injectable } from '@nestjs/common';
import { isDeepStrictEqual } from 'node:util';

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

import { PrismaService } from '../../../../prisma/prisma.service';

const FIRST_BATCH_OFFSET = 0;

export type ForecastSourceVersion = {
  dataRevision: number;
  fetchedAt: Date;
  instrumentUid: string;
  kind: string;
  rangeKey: string;
  source: string;
};
const forecastSourceReadBatchSize = 100;

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

  resolve(input: {
    candidate?: object;
    membershipRevision: number;
    portfolioProjection: string;
    portfolios: { accountId: string; dataRevision: number }[];
    revision: string;
    scopeKey: string;
    sources: ForecastSourceVersion[];
    timestamps: { at: string; resourceKey: string }[];
    userId: string;
  }) {
    return this.prisma.$transaction(async (tx) => {
      const owner = { userId: input.userId, revision: input.revision };
      if (input.candidate) {
        const current = await tx.tbankConnection.updateMany({
          where: {
            ...owner,
            membershipRevision: input.membershipRevision,
            invalidToken: false,
          },
          data: { membershipRevision: { increment: 0 } },
        });
        if (!current.count) return null;
      } else if (!(await tx.tbankConnection.findFirst({
        where: { ...owner, membershipRevision: input.membershipRevision, invalidToken: false },
      }))) return null;
      const key = { ...owner, scopeKey: input.scopeKey };
      const previous = await tx.tbankForecast.findUnique({
        where: { userId_revision_scopeKey: key },
      });
      const requestedWindow = eventsWindow(input.scopeKey);
      const saved = previous ?? (requestedWindow
        ? (
            await tx.tbankForecast.findMany({
              where: {
                ...owner,
                scopeKey: { startsWith: 'events/' },
                membershipRevision: input.membershipRevision,
              },
            })
          ).find((forecast) => {
            const window = eventsWindow(forecast.scopeKey);
            return (
              window &&
              window.accountId === requestedWindow.accountId &&
              window.from <= requestedWindow.to &&
              requestedWindow.from <= window.to
            );
          })
        : null);
      const fallback =
        saved?.membershipRevision === input.membershipRevision
          ? saved
          : null;
      if (!input.candidate) return fallback;
      if (!previous && fallback) return fallback;
      const portfolios = await tx.tbankPortfolio.findMany({
        where: {
          ...owner,
          accountId: { in: input.portfolios.map((row) => row.accountId) },
        },
        select: { accountId: true, dataRevision: true },
      });
      if (
        input.portfolios.some(
          (expected) =>
            !portfolios.some(
              (row) =>
                row.accountId === expected.accountId &&
                row.dataRevision === expected.dataRevision,
            ),
        )
      ) {
        return fallback;
      }
      // Four identity parameters per source: stay below SQLite's parameter limit.
      const sources: ForecastSourceVersion[] = [];
      for (
        let offset = FIRST_BATCH_OFFSET;
        offset < input.sources.length;
        offset += forecastSourceReadBatchSize
      ) {
        sources.push(
          ...(await tx.tbankInstrumentSource.findMany({
            where: {
              provider: 'tbank',
              OR: input.sources
                .slice(offset, offset + forecastSourceReadBatchSize)
                .map(({ instrumentUid, source, kind, rangeKey }) => ({
                  instrumentUid,
                  source,
                  kind,
                  rangeKey,
                })),
            },
            select: {
              instrumentUid: true,
              source: true,
              kind: true,
              rangeKey: true,
              dataRevision: true,
              fetchedAt: true,
            },
          })),
        );
      }
      if (
        input.sources.some(
          (expected) =>
            !sources.some(
              (row) =>
                row.instrumentUid === expected.instrumentUid &&
                row.source === expected.source &&
                row.kind === expected.kind &&
                row.rangeKey === expected.rangeKey &&
                row.dataRevision === expected.dataRevision &&
                row.fetchedAt.valueOf() === expected.fetchedAt.valueOf(),
            ),
        )
      ) {
        return fallback;
      }
      const revisions = JSON.parse(
        JSON.stringify({
          portfolioProjection: input.portfolioProjection,
          portfolios: input.portfolios,
          sources: input.sources,
        }),
      ) as Prisma.InputJsonObject;
      const timestamps = input.timestamps as Prisma.InputJsonArray;
      if (
        fallback &&
        isDeepStrictEqual(fallback.dependencyRevisions, revisions) &&
        isDeepStrictEqual(fallback.sourceTimestamps, timestamps)
      ) {
        return fallback;
      }
      const data = {
        membershipRevision: input.membershipRevision,
        payload: JSON.parse(
          JSON.stringify(input.candidate),
        ) as Prisma.InputJsonObject,
        dependencyRevisions: revisions,
        sourceTimestamps: timestamps,
        calculatedAt: new Date(),
      };
      return tx.tbankForecast.upsert({
        where: { userId_revision_scopeKey: key },
        create: { ...key, ...data },
        update: { ...data, dataRevision: { increment: 1 } },
      });
    });
  }
}

function eventsWindow(scopeKey: string) {
  if (!scopeKey.startsWith('events/')) return null;
  try {
    const [accountId, from, to] = JSON.parse(scopeKey.slice('events/'.length)) as [string, string, string];
    return { accountId, from, to };
  } catch {
    return null;
  }
}