import { BadRequestException, Injectable, NotFoundException, } from '@nestjs/common'; import { createHash } from 'node:crypto'; import { PrismaService } from '../../../../prisma/prisma.service'; import { dateOnlyBounds, DateOnlyRange, moscowDate, } from '../../investment-calendar'; import { OperationErrorCode } from '../domain/operation-error-code'; import { decodeInterval } from '../domain/operations-interval-codec'; const CURSOR_LOOKAHEAD_ROWS = 1; @Injectable() export class BrokerageOperationsReadRepository { constructor(private readonly prisma: PrismaService) {} async readPage(input: { accountId: string; categories: string[]; cursor?: { accountId: string; at: string; binding: string; id: string }; filterKey: string; from: Date; instrumentUid?: string; limit?: number; periods?: (openedAt: Date | undefined) => DateOnlyRange[]; revision: string; to: Date; userId: string; }) { return this.prisma.$transaction(async (tx) => { const connection = await tx.tbankConnection.findUnique({ where: { userId_revision: { userId: input.userId, revision: input.revision }, invalidToken: false, }, select: { membershipAt: true, membershipRevision: true, accounts: { where: input.accountId === 'all' ? {} : { accountId: input.accountId }, orderBy: { accountId: 'asc' }, select: { accountId: true, name: true, openedOn: true, operationsRevision: true, }, }, }, }); if (!connection) return null; if ( connection.membershipAt && input.accountId !== 'all' && !connection.accounts.length ) { throw new NotFoundException(); } const accounts = connection.accounts; // eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings. const oldest = accounts.map((account) => account.openedOn).sort()[0]; const openedAt = oldest && accounts.every((account) => account.openedOn) ? new Date(`${oldest}T00:00:00+03:00`) : undefined; const periods = input.periods?.(openedAt).map((period) => { const now = new Date(); const bounds = dateOnlyBounds(period, now); return { ...bounds, key: `${period.from}/${period.to}`, movingEnd: period.to >= moscowDate(now), }; }); const binding = createHash('sha256') .update( JSON.stringify([ input.userId, input.revision, input.accountId === 'all' ? connection.membershipRevision : null, accounts.map((account) => [ account.accountId, account.operationsRevision, ]), input.filterKey, ]), ) .digest('base64url'); if (input.cursor && input.cursor.binding !== binding) { throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor }); } const scope = { userId: input.userId, revision: input.revision, accountId: { in: accounts.map((account) => account.accountId) }, }; const coverage = await tx.tbankOperationCoverage.findMany({ where: scope, }); const work = await tx.tbankSyncWork.findMany({ where: { userId: input.userId, revision: input.revision, scopeKey: { startsWith: 'operations:' }, state: { in: ['running', 'failed'] }, }, select: { id: true, state: true, errorCode: true, intent: true, priority: true, }, }); const membershipWork = await tx.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: { userId: input.userId, revision: input.revision, scopeKey: 'membership', }, }, select: { id: true, state: true, errorCode: true }, }); const after = input.cursor; const rows = await tx.tbankOperation.findMany({ where: { ...scope, ...(periods ? { OR: periods.map((period) => ({ occurredAt: { gte: period.from, lt: period.to }, })), } : { occurredAt: { gte: input.from, lt: input.to } }), ...(input.instrumentUid ? { instrumentUid: input.instrumentUid } : {}), ...(input.categories.length ? { category: { in: input.categories } } : {}), ...(after ? { OR: [ { occurredAt: { lt: new Date(after.at) } }, { occurredAt: new Date(after.at), accountId: { gt: after.accountId }, }, { occurredAt: new Date(after.at), accountId: after.accountId, brokerOperationId: { gt: after.id }, }, ], } : {}), }, orderBy: [ { occurredAt: 'desc' }, { accountId: 'asc' }, { brokerOperationId: 'asc' }, ], ...(input.limit === undefined ? {} : { take: input.limit + CURSOR_LOOKAHEAD_ROWS }), }); return { connection, periods, membershipWork, binding, rows, coverage: coverage.map(({ instrumentPredicate, ...row }) => ({ ...row, instrumentUid: instrumentPredicate || undefined, })), work: work.map(({ intent, ...row }) => ({ ...row, intervals: (intent as { intervals: string[] }).intervals.map( decodeInterval, ), })), }; }); } }