From 5aa52586fdf1d52762c9fe05a697a9d3805a5cc8 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Wed, 4 Nov 2026 14:02:28 +0300 Subject: [PATCH] refactor: split brokerage operations repository and cursor codec (WealthLedger/backend#167) --- eslint-suppressions.json | 2 +- .../brokerage-operations.service.ts | 209 +++------------ .../local-operations-cursor.codec.spec.ts | 23 ++ .../local-operations-cursor.codec.ts | 72 +++++ .../operations/domain/operations-coverage.ts | 165 ++++++++++++ .../domain/operations-interval-codec.ts | 67 +++++ .../brokerage-operations-read.repository.ts | 192 ++++++++++++++ .../brokerage-operations.repository.ts | 251 +----------------- .../operations/operations.module.ts | 4 + 9 files changed, 570 insertions(+), 415 deletions(-) create mode 100644 src/modules/investments/operations/application/local-operations-cursor.codec.spec.ts create mode 100644 src/modules/investments/operations/application/local-operations-cursor.codec.ts create mode 100644 src/modules/investments/operations/domain/operations-coverage.ts create mode 100644 src/modules/investments/operations/domain/operations-interval-codec.ts create mode 100644 src/modules/investments/operations/infrastructure/brokerage-operations-read.repository.ts diff --git a/eslint-suppressions.json b/eslint-suppressions.json index af6f170..80fdc16 100644 --- a/eslint-suppressions.json +++ b/eslint-suppressions.json @@ -101,7 +101,7 @@ "count": 2 }, "sonarjs/no-nested-conditional": { - "count": 9 + "count": 6 } }, "src/modules/investments/operations/domain/operations-sync-plan.ts": { diff --git a/src/modules/investments/operations/application/brokerage-operations.service.ts b/src/modules/investments/operations/application/brokerage-operations.service.ts index fcabee2..3b7c2c8 100644 --- a/src/modules/investments/operations/application/brokerage-operations.service.ts +++ b/src/modules/investments/operations/application/brokerage-operations.service.ts @@ -1,6 +1,5 @@ -import { BadRequestException, Injectable } from '@nestjs/common'; -import { ConfigService } from '@nestjs/config'; -import { createHash, createHmac, timingSafeEqual } from 'node:crypto'; +import { Injectable } from '@nestjs/common'; +import { createHash } from 'node:crypto'; import { InstrumentMetadataService, @@ -11,51 +10,47 @@ import { } from '../../../../integrations/tbank/tbank.api'; import { dateOnlyBounds, DateOnlyRange } from '../../investment-calendar'; import { investmentResourceMeta } from '../../investment-resource.mapper'; -import { accountHistory, ledgerRange } from '../../investment-sync-plan'; +import { ledgerRange } from '../../investment-sync-plan'; import { hourlyAttempt, TbankSyncRepository, } from '../../tbank-sync/tbank-sync.api'; -import { OperationErrorCode } from '../domain/operation-error-code'; +import { + coverageAsOf, + deriveReadCoverage, + historyState, + serializeRange, +} from '../domain/operations-coverage'; import { confirmedOperationRanges, operationIntervalsOverlap, - operationsSyncPriority, planOperationRead, uncoveredOperationIntervals, } from '../domain/operations-sync-plan'; import { mapBrokerageOperation } from '../infrastructure/brokerage-operation.mapper'; -import { BrokerageOperationsRepository } from '../infrastructure/brokerage-operations.repository'; +import { BrokerageOperationsReadRepository } from '../infrastructure/brokerage-operations-read.repository'; import { BrokerageOperationsQueryDto, DEFAULT_OPERATIONS_PAGE_SIZE, } from '../presentation/brokerage-operation.dto'; import { BrokerageOperationsSyncService } from './brokerage-operations-sync.service'; +import { LocalOperationsCursorCodec } from './local-operations-cursor.codec'; const FIRST_RESULT_OFFSET = 0; const MISSING_MEMBERSHIP_REVISION = 0; const STALE_FRESHNESS_MS = 0; -const UNKNOWN_AS_OF_MS = 0; +const NO_COVERAGE_ROWS = 0; const LAST_ROW_INDEX = -1; const UNIX_EPOCH_MS = 0; -type LocalOperationsCursor = { - accountId: string; - at: string; - binding: string; - id: string; - movingEnd?: boolean; - to: string; -}; - @Injectable() export class BrokerageOperationsService { constructor( private readonly access: TbankAccessService, private readonly metadata: InstrumentMetadataService, - private readonly repository: BrokerageOperationsRepository, + private readonly repository: BrokerageOperationsReadRepository, private readonly sync: BrokerageOperationsSyncService, - private readonly config: ConfigService, + private readonly cursorCodec: LocalOperationsCursorCodec, private readonly hourlySync: TbankSyncRepository, ) {} @@ -65,7 +60,7 @@ export class BrokerageOperationsService { query: BrokerageOperationsQueryDto, ) { const connection = await this.access.local(userId, accountId); - const cursor = query.cursor ? this.decodeCursor(query.cursor) : undefined; + const cursor = query.cursor ? this.cursorCodec.decode(query.cursor) : undefined; const selected = connection.accounts.filter( (account) => accountId === 'all' || account.accountId === accountId, ); @@ -124,64 +119,18 @@ export class BrokerageOperationsService { const saved = await this.repository.readPage(readInput); const hourly = await this.hourlySync.hourly(userId); const accounts = saved?.connection.accounts ?? []; - const targets = accounts.flatMap((account) => - from < to - ? [ - { - accountId: account.accountId, - from, - to, - instrumentUid: query.instrumentUid, - }, - ] - : [], - ); - const plan = planOperationRead(targets, saved?.coverage ?? [], now, { + const view = deriveReadCoverage({ + accounts, + connectionMembershipAt: saved?.connection.membershipAt ?? null, + coverage: saved?.coverage ?? [], + from, + instrumentUid: query.instrumentUid, movingEnd, - }); - const confirmed = confirmedOperationRanges(plan.intervals, saved?.coverage ?? []); - const missing = uncoveredOperationIntervals( - plan.intervals, - confirmed.map((range) => ({ - ...range, - instrumentUid: query.instrumentUid, - })), - ); - const available = Boolean(saved?.connection.membershipAt); - const complete = available && !missing.length; - const relevant = - saved?.work.filter((work) => - work.intervals.some((interval) => - targets.some((target) => operationIntervalsOverlap(interval, target)), - ), - ) ?? []; - const work = - relevant.find((work) => work.state === 'running') ?? - relevant.find((work) => work.state === 'failed') ?? - null; - const sources = (saved?.coverage ?? []).filter((row) => - targets.some( - (target) => - operationIntervalsOverlap(row, target) && - (!row.instrumentUid || row.instrumentUid === target.instrumentUid), - ), - ); - const lastSuccessfulAt = - sources - .map((row) => row.fetchedAt) - .sort((a, b) => a.valueOf() - b.valueOf())[0] ?? - (complete ? saved?.connection.membershipAt : null) ?? - null; - const historyTargets = accountHistory(accounts, now); - const history = planOperationRead( - historyTargets, - saved?.coverage ?? [], now, - { movingEnd: true }, - ); - const knownBoundary = accounts.every( - (account) => account.openedOn !== null, - ); + to, + work: saved?.work ?? [], + }); + const { targets, plan, confirmed, missing, available, complete, lastSuccessfulAt, history, knownBoundary, work } = view; const rows = saved?.rows.slice(FIRST_RESULT_OFFSET, limit) ?? []; const instruments = await this.metadata.savedMany( rows.map((row) => row.payload as TbankOperationItem), @@ -224,49 +173,17 @@ export class BrokerageOperationsService { lastAttempt: hourlyAttempt(hourly, 'operations', accountId === 'all' ? undefined : accountId), dependencies: [membershipMeta, ...metadataDependencies], coverage: { - requested: targets.map(serialize), - completeRanges: confirmed.map(serialize), - missingRanges: missing.map(serialize), - asOf: - accounts.length && - accounts.every((account) => - plan.intervals.some( - (interval) => - interval.accountId === account.accountId && interval.asOf, - ), - ) - ? new Date( - Math.min( - ...accounts.map((account) => - Math.max( - ...plan.intervals - .filter( - (interval) => - interval.accountId === account.accountId, - ) - .map((interval) => interval.asOf?.valueOf() ?? UNKNOWN_AS_OF_MS), - ), - ), - ), - ).toISOString() - : null, + requested: targets.map(serializeRange), + completeRanges: confirmed.map(serializeRange), + missingRanges: missing.map(serializeRange), + asOf: coverageAsOf(plan, accounts.map((account) => account.accountId)), historyBoundary: knownBoundary ? 'known' : 'unknown', - historyState: saved?.work.some( - (work) => work.state === 'running' && work.priority > operationsSyncPriority.screen, - ) - ? 'loading' - : saved?.work.some( - (work) => work.state === 'failed' && work.priority > operationsSyncPriority.screen, - ) - ? 'failed' - : knownBoundary && - history.intervals.every( - (interval) => interval.availability === 'complete', - ) - ? 'complete' - : saved?.coverage.length - ? 'partial' - : 'missing', + historyState: historyState( + knownBoundary, + history, + saved?.work ?? [], + saved?.coverage.length ?? NO_COVERAGE_ROWS, + ), }, }); const last = rows.at(LAST_ROW_INDEX); @@ -301,7 +218,7 @@ export class BrokerageOperationsService { hasNext, nextCursor: hasNext && last && saved - ? this.encodeCursor({ + ? this.cursorCodec.encode({ binding: saved.binding, movingEnd, to: to.toISOString(), @@ -316,50 +233,6 @@ export class BrokerageOperationsService { }; } - private encodeCursor(value: LocalOperationsCursor): string { - const body = Buffer.from(JSON.stringify(value)).toString('base64url'); - return `${body}.${createHmac('sha256', this.config.getOrThrow('JWT_SECRET')).update(body).digest('base64url')}`; - } - - private decodeCursor(value: string): LocalOperationsCursor { - try { - const [body, signature, extra] = value.split('.'); - const expected = createHmac( - 'sha256', - this.config.getOrThrow('JWT_SECRET'), - ) - .update(body) - .digest(); - const actual = Buffer.from(signature ?? '', 'base64url'); - if ( - extra || - actual.length !== expected.length || - !timingSafeEqual(actual, expected) - ) { - throw new Error(); - } - const cursor = JSON.parse( - Buffer.from(body, 'base64url').toString(), - ) as Partial; - if ( - (cursor.movingEnd !== undefined && - typeof cursor.movingEnd !== 'boolean') || - !cursor.binding || - !cursor.accountId || - !cursor.id || - !cursor.at || - !cursor.to || - !Number.isFinite(Date.parse(cursor.at)) || - !Number.isFinite(Date.parse(cursor.to)) - ) { - throw new Error(); - } - return cursor as LocalOperationsCursor; - } catch { - throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor }); - } - } - async collectRanges( userId: string, accountId: string, @@ -466,12 +339,12 @@ export class BrokerageOperationsService { lastAttempt: hourlyAttempt(hourly, 'operations', accountId === 'all' ? undefined : accountId), dependencies: [membership], coverage: { - requested: targets.map(serialize), - completeRanges: confirmed.map(serialize), + requested: targets.map(serializeRange), + completeRanges: confirmed.map(serializeRange), missingRanges: uncoveredOperationIntervals( plan.intervals, confirmed, - ).map(serialize), + ).map(serializeRange), asOf: complete && targets.length ? new Date( @@ -541,7 +414,3 @@ export class BrokerageOperationsService { }); } } - -function serialize(range: { accountId: string; from: Date; to: Date }) { - return { accountId: range.accountId, from: range.from.toISOString(), to: range.to.toISOString() }; -} diff --git a/src/modules/investments/operations/application/local-operations-cursor.codec.spec.ts b/src/modules/investments/operations/application/local-operations-cursor.codec.spec.ts new file mode 100644 index 0000000..02c6c8f --- /dev/null +++ b/src/modules/investments/operations/application/local-operations-cursor.codec.spec.ts @@ -0,0 +1,23 @@ +import { decode, encode } from './local-operations-cursor.codec'; + +const secret = 'secret'; +const cursor = { + accountId: 'account', + at: '2026-01-01T00:00:00.000Z', + binding: 'owner-filter-binding', + id: 'operation', + movingEnd: false, + to: '2026-02-01T00:00:00.000Z', +}; + +describe('local operations cursor', () => { + it('roundtrips the opaque cursor, including its owner and filter binding', () => { + expect(decode(encode(cursor, secret), secret)).toEqual(cursor); + }); + + it('rejects a tampered cursor and a cursor signed with a different key', () => { + const encoded = encode(cursor, secret); + expect(() => decode(`${encoded}x`, secret)).toThrow(); + expect(() => decode(encoded, 'different-secret')).toThrow(); + }); +}); diff --git a/src/modules/investments/operations/application/local-operations-cursor.codec.ts b/src/modules/investments/operations/application/local-operations-cursor.codec.ts new file mode 100644 index 0000000..9445bcf --- /dev/null +++ b/src/modules/investments/operations/application/local-operations-cursor.codec.ts @@ -0,0 +1,72 @@ +import { BadRequestException, Injectable } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { createHmac, timingSafeEqual } from 'node:crypto'; + +import { OperationErrorCode } from '../domain/operation-error-code'; + +export type LocalOperationsCursor = { + accountId: string; + at: string; + binding: string; + id: string; + movingEnd?: boolean; + to: string; +}; + +// Opaque cursor: base64url body + HMAC-SHA256 signature over JWT_SECRET. +// Wire format and validation are part of the HTTP contract and must not change. +export function encode(value: LocalOperationsCursor, secret: string): string { + const body = Buffer.from(JSON.stringify(value)).toString('base64url'); + return `${body}.${createHmac('sha256', secret).update(body).digest('base64url')}`; +} + +export function decode(value: string, secret: string): LocalOperationsCursor { + try { + const [body, signature, extra] = value.split('.'); + const expected = createHmac('sha256', secret).update(body).digest(); + const actual = Buffer.from(signature ?? '', 'base64url'); + if ( + extra || + actual.length !== expected.length || + !timingSafeEqual(actual, expected) + ) { + throw new Error(); + } + const cursor = JSON.parse( + Buffer.from(body, 'base64url').toString(), + ) as Partial; + if ( + (cursor.movingEnd !== undefined && + typeof cursor.movingEnd !== 'boolean') || + !cursor.binding || + !cursor.accountId || + !cursor.id || + !cursor.at || + !cursor.to || + !Number.isFinite(Date.parse(cursor.at)) || + !Number.isFinite(Date.parse(cursor.to)) + ) { + throw new Error(); + } + return cursor as LocalOperationsCursor; + } catch { + throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor }); + } +} + +@Injectable() +export class LocalOperationsCursorCodec { + constructor(private readonly config: ConfigService) {} + + encode(value: LocalOperationsCursor): string { + return encode(value, this.secret()); + } + + decode(value: string): LocalOperationsCursor { + return decode(value, this.secret()); + } + + private secret(): string { + return this.config.getOrThrow('JWT_SECRET'); + } +} diff --git a/src/modules/investments/operations/domain/operations-coverage.ts b/src/modules/investments/operations/domain/operations-coverage.ts new file mode 100644 index 0000000..0029f7d --- /dev/null +++ b/src/modules/investments/operations/domain/operations-coverage.ts @@ -0,0 +1,165 @@ +import { accountHistory } from '../../investment-sync-plan'; +import { confirmedOperationRanges, type OperationCoverage, + type OperationInterval, + operationIntervalsOverlap, + operationsSyncPriority, + planOperationRead, + uncoveredOperationIntervals } from './operations-sync-plan'; + +const UNKNOWN_AS_OF_MS = 0; + +type WorkState = { + errorCode: null | string; + id: string; + intervals: OperationInterval[]; + priority: number; + state: string; +}; + +type CoverageAccount = { accountId: string; openedOn: null | string }; + +// Pure derivation of the ledger read view: what was requested, what is +// confirmed, what is missing, and which sync work is currently relevant. +export function deriveReadCoverage(input: { + accounts: CoverageAccount[]; + connectionMembershipAt: Date | null; + coverage: OperationCoverage[]; + from: Date; + instrumentUid?: string; + movingEnd: boolean; + now: Date; + to: Date; + work: WorkState[]; +}) { + const targets = input.accounts.flatMap((account) => + input.from < input.to + ? [ + { + accountId: account.accountId, + from: input.from, + to: input.to, + instrumentUid: input.instrumentUid, + }, + ] + : [], + ); + const plan = planOperationRead(targets, input.coverage, input.now, { + movingEnd: input.movingEnd, + }); + const confirmed = confirmedOperationRanges(plan.intervals, input.coverage); + const missing = uncoveredOperationIntervals( + plan.intervals, + confirmed.map((range) => ({ + ...range, + instrumentUid: input.instrumentUid, + })), + ); + const available = Boolean(input.connectionMembershipAt); + const complete = available && !missing.length; + const relevant = input.work.filter((work) => + work.intervals.some((interval) => + targets.some((target) => operationIntervalsOverlap(interval, target)), + ), + ); + const work = + relevant.find((work) => work.state === 'running') ?? + relevant.find((work) => work.state === 'failed') ?? + null; + const sources = input.coverage.filter((row) => + targets.some( + (target) => + operationIntervalsOverlap(row, target) && + (!row.instrumentUid || row.instrumentUid === target.instrumentUid), + ), + ); + const lastSuccessfulAt = + sources + .map((row) => row.fetchedAt) + .sort((a, b) => a.valueOf() - b.valueOf())[0] ?? + (complete ? input.connectionMembershipAt : null) ?? + null; + const history = planOperationRead( + accountHistory(input.accounts, input.now), + input.coverage, + input.now, + { movingEnd: true }, + ); + const knownBoundary = input.accounts.every( + (account) => account.openedOn !== null, + ); + return { + targets, + plan, + confirmed, + missing, + available, + complete, + work, + lastSuccessfulAt, + history, + knownBoundary, + }; +} + +export function coverageAsOf( + plan: ReturnType, + accountIds: string[], +): null | string { + if ( + accountIds.length && + accountIds.every((accountId) => + plan.intervals.some( + (interval) => + interval.accountId === accountId && interval.asOf, + ), + ) + ) { + return new Date( + Math.min( + ...accountIds.map((accountId) => + Math.max( + ...plan.intervals + .filter((interval) => interval.accountId === accountId) + .map((interval) => interval.asOf?.valueOf() ?? UNKNOWN_AS_OF_MS), + ), + ), + ), + ).toISOString(); + } + return null; +} + +export function historyState( + knownBoundary: boolean, + history: ReturnType, + work: WorkState[], + coverageCount: number, +): 'complete' | 'failed' | 'loading' | 'missing' | 'partial' { + if ( + work.some( + (work) => work.state === 'running' && work.priority > operationsSyncPriority.screen, + ) + ) { + return 'loading'; + } + if ( + work.some( + (work) => work.state === 'failed' && work.priority > operationsSyncPriority.screen, + ) + ) { + return 'failed'; + } + if ( + knownBoundary && + history.intervals.every( + (interval) => interval.availability === 'complete', + ) + ) { + return 'complete'; + } + return coverageCount ? 'partial' : 'missing'; +} + +export function serializeRange(range: { accountId: string; from: Date; to: Date }) { + return { accountId: range.accountId, from: range.from.toISOString(), to: range.to.toISOString() }; +} diff --git a/src/modules/investments/operations/domain/operations-interval-codec.ts b/src/modules/investments/operations/domain/operations-interval-codec.ts new file mode 100644 index 0000000..edf8b61 --- /dev/null +++ b/src/modules/investments/operations/domain/operations-interval-codec.ts @@ -0,0 +1,67 @@ +import { type OperationInterval, + operationIntervalsOverlap, + operationsSyncPriority } from './operations-sync-plan'; + +export type OperationsIntent = { + atomic: boolean; + atomicIntervals?: string[]; + intervals: string[]; + screenIntervals?: string[]; +}; + +export function intervalKey(interval: OperationInterval): string { + // eslint-disable-next-line sonarjs/no-inverted-boolean-check -- Negation also rejects Invalid Date; >= would silently accept NaN. + if (!(interval.from < interval.to)) { + throw new Error('Invalid operations interval'); + } + return JSON.stringify([ + interval.accountId, + interval.from.toISOString(), + interval.to.toISOString(), + interval.instrumentUid ?? '', + ]); +} + +export function decodeInterval(key: string): OperationInterval { + const [accountId, from, to, instrumentUid] = JSON.parse(key) as string[]; + return { + accountId, + from: new Date(from), + to: new Date(to), + ...(instrumentUid ? { instrumentUid } : {}), + }; +} + +export function extendRunningIntent(input: { + additions: OperationInterval[]; + intervals: OperationInterval[]; + pending: OperationInterval[]; + previous: OperationsIntent | undefined; + previousPriority: number; + priority: number; + publicationAtomic: boolean; +}): OperationsIntent { + const { previous, pending, additions, intervals, priority, previousPriority, publicationAtomic } = input; + const overlapsRequested = (item: OperationInterval) => + intervals.some((target) => operationIntervalsOverlap(item, target)); + const retainedScreen = previous?.screenIntervals ?? + (previousPriority === operationsSyncPriority.screen ? (previous?.intervals ?? []) : []); + const addedScreen = priority === operationsSyncPriority.screen + ? [...pending, ...additions].filter(overlapsRequested).map(intervalKey) + : []; + const ordered = priority < previousPriority + ? [ + ...pending.filter(overlapsRequested), + ...additions, + ...pending.filter((item) => !overlapsRequested(item)), + ] + : [...pending, ...additions]; + return { + atomic: previous?.atomic ?? publicationAtomic, + atomicIntervals: previous?.atomic + ? [...new Set([...(previous.atomicIntervals ?? previous.intervals), ...additions.map(intervalKey)])] + : [], + screenIntervals: [...new Set([...retainedScreen, ...addedScreen])], + intervals: ordered.map(intervalKey), + }; +} diff --git a/src/modules/investments/operations/infrastructure/brokerage-operations-read.repository.ts b/src/modules/investments/operations/infrastructure/brokerage-operations-read.repository.ts new file mode 100644 index 0000000..1b4ef1d --- /dev/null +++ b/src/modules/investments/operations/infrastructure/brokerage-operations-read.repository.ts @@ -0,0 +1,192 @@ +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, + ), + })), + }; + }); + } +} diff --git a/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts b/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts index 844d1a8..c9819e1 100644 --- a/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts +++ b/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts @@ -1,8 +1,4 @@ -import { - BadRequestException, - Injectable, - NotFoundException, -} from '@nestjs/common'; +import { Injectable, NotFoundException } from '@nestjs/common'; import { createHash, randomUUID } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; @@ -16,11 +12,11 @@ import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code' import { isTbankProtocolValue, TbankOperationState } from '../../../../integrations/tbank/tbank-protocol'; import { PrismaService } from '../../../../prisma/prisma.service'; import { - dateOnlyBounds, - DateOnlyRange, - moscowDate, -} from '../../investment-calendar'; -import { OperationErrorCode } from '../domain/operation-error-code'; + decodeInterval, + extendRunningIntent, + intervalKey, + type OperationsIntent, +} from '../domain/operations-interval-codec'; import { type OperationInterval, operationIntervals, @@ -33,13 +29,13 @@ import { operationTimestamp, } from './brokerage-operation.mapper'; -const CURSOR_LOOKAHEAD_ROWS = 1; const EMPTY_INTERVAL_COUNT = 0; const SINGLE_WORK_GROUP = 1; const INITIAL_PAGE_COUNT = 0; const PAGE_INCREMENT = 1; const EXPECTED_UPDATED_WORK_ITEMS = 1; +export type { OperationsIntent } from '../domain/operations-interval-codec'; export type { OperationInterval } from '../domain/operations-sync-plan'; export type OperationAttempt = { @@ -48,12 +44,6 @@ export type OperationAttempt = { revision: string; userId: string; }; -type OperationsIntent = { - atomic: boolean; - atomicIntervals?: string[]; - intervals: string[]; - screenIntervals?: string[]; -}; type PageProgress = { complete: boolean; cursor: string; @@ -66,176 +56,6 @@ const progressItem = ''; export class BrokerageOperationsRepository { 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, - ), - })), - }; - }); - } - async coverage(userId: string, revision: string, accountIds: string[]) { const rows = await this.prisma.tbankOperationCoverage.findMany({ where: { @@ -882,53 +702,6 @@ export class BrokerageOperationsRepository { } } -function extendRunningIntent(input: { - additions: OperationInterval[]; - intervals: OperationInterval[]; - pending: OperationInterval[]; - previous: OperationsIntent | undefined; - previousPriority: number; - priority: number; - publicationAtomic: boolean; -}): OperationsIntent { - const { previous, pending, additions, intervals, priority, previousPriority, publicationAtomic } = input; - const overlapsRequested = (item: OperationInterval) => - intervals.some((target) => operationIntervalsOverlap(item, target)); - const retainedScreen = previous?.screenIntervals ?? - (previousPriority === operationsSyncPriority.screen ? (previous?.intervals ?? []) : []); - const addedScreen = priority === operationsSyncPriority.screen - ? [...pending, ...additions].filter(overlapsRequested).map(intervalKey) - : []; - const ordered = priority < previousPriority - ? [ - ...pending.filter(overlapsRequested), - ...additions, - ...pending.filter((item) => !overlapsRequested(item)), - ] - : [...pending, ...additions]; - return { - atomic: previous?.atomic ?? publicationAtomic, - atomicIntervals: previous?.atomic - ? [...new Set([...(previous.atomicIntervals ?? previous.intervals), ...additions.map(intervalKey)])] - : [], - screenIntervals: [...new Set([...retainedScreen, ...addedScreen])], - intervals: ordered.map(intervalKey), - }; -} - -function intervalKey(interval: OperationInterval): string { - // eslint-disable-next-line sonarjs/no-inverted-boolean-check -- Negation also rejects Invalid Date; >= would silently accept NaN. - if (!(interval.from < interval.to)) { - throw new Error('Invalid operations interval'); - } - return JSON.stringify([ - interval.accountId, - interval.from.toISOString(), - interval.to.toISOString(), - interval.instrumentUid ?? '', - ]); -} - async function claim( tx: Prisma.TransactionClient, attempt: OperationAttempt, @@ -943,13 +716,3 @@ async function claim( }); return result.count === EXPECTED_UPDATED_WORK_ITEMS; } - -function decodeInterval(key: string): OperationInterval { - const [accountId, from, to, instrumentUid] = JSON.parse(key) as string[]; - return { - accountId, - from: new Date(from), - to: new Date(to), - ...(instrumentUid ? { instrumentUid } : {}), - }; -} diff --git a/src/modules/investments/operations/operations.module.ts b/src/modules/investments/operations/operations.module.ts index d62f1ad..ecb5b47 100644 --- a/src/modules/investments/operations/operations.module.ts +++ b/src/modules/investments/operations/operations.module.ts @@ -5,6 +5,8 @@ import { AuthModule } from '../../auth/auth.module'; import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module'; import { BrokerageOperationsSyncService } from './application/brokerage-operations-sync.service'; import { BrokerageOperationsService } from './application/brokerage-operations.service'; +import { LocalOperationsCursorCodec } from './application/local-operations-cursor.codec'; +import { BrokerageOperationsReadRepository } from './infrastructure/brokerage-operations-read.repository'; import { BrokerageOperationsRepository } from './infrastructure/brokerage-operations.repository'; import { BrokerageOperationsController } from './presentation/brokerage-operations.controller'; @@ -14,6 +16,8 @@ import { BrokerageOperationsController } from './presentation/brokerage-operatio providers: [ BrokerageOperationsService, BrokerageOperationsRepository, + BrokerageOperationsReadRepository, + LocalOperationsCursorCodec, BrokerageOperationsSyncService, ], exports: [BrokerageOperationsService, BrokerageOperationsSyncService], -- 2.54.0