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 '#src/prisma/prisma.service';
import { toJson } from '#src/prisma/to-json';

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 }[];
    replacesLatest?: boolean;
    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 },
      });
      // An events read serves the account's latest complete forecast whatever
      // the requested window; other scopes keep their exact-key row.
      const requestedWindow = eventsWindow(input.scopeKey);
      const latest = requestedWindow
        ? await tx.tbankForecast.findFirst({
            where: {
              ...owner,
              // scopeKey is `events/[accountId,from,to]` JSON: pin the account.
              scopeKey: {
                startsWith: `events/[${JSON.stringify(requestedWindow.accountId)},`,
              },
              membershipRevision: input.membershipRevision,
            },
            orderBy: [{ calculatedAt: 'desc' }, { scopeKey: 'asc' }],
          })
        : previous;
      const fallback =
        latest?.membershipRevision === input.membershipRevision ? latest : null;
      if (!input.candidate) return fallback;
      // Only the default window may replace the latest forecast; an explicit
      // range is served from it and publishes only when nothing is saved.
      if (!input.replacesLatest && fallback && fallback.scopeKey !== input.scopeKey) {
        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 = toJson({
        portfolioProjection: input.portfolioProjection,
        portfolios: input.portfolios,
        sources: input.sources,
      });
      const timestamps = input.timestamps as Prisma.InputJsonArray;
      if (
        fallback?.scopeKey === input.scopeKey &&
        isDeepStrictEqual(fallback.dependencyRevisions, revisions) &&
        isDeepStrictEqual(fallback.sourceTimestamps, timestamps)
      ) {
        return fallback;
      }
      const data = {
        membershipRevision: input.membershipRevision,
        payload: toJson(input.candidate),
        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;
  }
}