import { BadRequestException, Injectable } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { createHash, createHmac, timingSafeEqual } from 'node:crypto'; import { InstrumentMetadataService, metadataKey, TbankAccessService, TbankMembershipService, type TbankOperationItem, tbankSourceTtlMs, } from '../../../../integrations/tbank/tbank.api'; import { dateOnlyBounds, DateOnlyRange } from '../../investment-calendar'; import { investmentResourceMeta } from '../../investment-resource.mapper'; import { accountHistory, ledgerRange, planInvestmentScreen, } from '../../investment-sync-plan'; import { hourlyAttempt, TbankSyncRepository, } from '../../tbank-sync/tbank-sync.repository'; import { OperationErrorCode } from '../domain/operation-error-code'; 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 { BrokerageOperationsQueryDto, DEFAULT_OPERATIONS_PAGE_SIZE, } from '../presentation/brokerage-operation.dto'; import { BrokerageOperationsSyncService } from './brokerage-operations-sync.service'; const EXCLUSIVE_END_OFFSET_MS = 1; const FIRST_RESULT_OFFSET = 0; const MISSING_MEMBERSHIP_REVISION = 0; const STALE_FRESHNESS_MS = 0; const UNKNOWN_AS_OF_MS = 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 sync: BrokerageOperationsSyncService, private readonly membership: TbankMembershipService, private readonly config: ConfigService, private readonly hourlySync: TbankSyncRepository, ) {} async read( userId: string, accountId: string, query: BrokerageOperationsQueryDto, ) { const connection = await this.access.local(userId, accountId); const cursor = query.cursor ? this.decodeCursor(query.cursor) : undefined; const selected = connection.accounts.filter( (account) => accountId === 'all' || account.accountId === accountId, ); const now = new Date(); const range = ledgerRange( query.from, query.to, query.instrumentUid, selected, now, ); const from = range.from; const to = cursor ? new Date(cursor.to) : range.to; const movingEnd = cursor?.movingEnd ?? range.to.valueOf() === now.valueOf(); // 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 categories = [...new Set(query.categories ?? [])].sort(); const limit = query.limit ?? DEFAULT_OPERATIONS_PAGE_SIZE; const filterKey = JSON.stringify([ accountId, from.toISOString(), query.to ? new Date( /^\d{4}-\d{2}-\d{2}$/.test(query.to) ? `${query.to}T23:59:59.999Z` : query.to, ).toISOString() : 'now', query.instrumentUid ?? null, categories, limit, ]); const readInput = { userId, revision: connection.revision, accountId, from, to, instrumentUid: query.instrumentUid, categories, limit, filterKey, cursor, }; if (cursor) await this.repository.readPage(readInput); const screen = planInvestmentScreen( { screen: 'operations', accountId, from: from.toISOString(), to: new Date(to.valueOf() - EXCLUSIVE_END_OFFSET_MS).toISOString(), }, { accounts: selected, instruments: [], membershipKnown: Boolean(connection.membershipAt), }, now, ); const intervals = screen.sources.flatMap((source) => source.kind === 'operations' && source.priority === 'screen' && source.accountId && source.from && source.to ? [{ accountId: source.accountId, from: source.from, to: source.to }] : [], ); // The ledger only starts the first-time load of missing intervals; the // mutable tail belongs to the hourly walk and membership is not rechecked. if (intervals.length) { await this.sync.readPlan(userId, intervals, { movingEnd, history: true, revision: connection.revision, }); } 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, { 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, ); const rows = saved?.rows.slice(FIRST_RESULT_OFFSET, limit) ?? []; const instruments = await this.metadata.savedMany( rows.map((row) => row.payload as TbankOperationItem), ); const metadataDependencies = [...instruments] .filter(([, value]) => value.complete) .map(([key, value]) => investmentResourceMeta({ resourceKey: `metadata/${key}`, dataRevision: value.dataRevision, availability: 'complete', fetchedAt: value.fetchedAt, freshForMs: tbankSourceTtlMs.portfolio, work: null, }), ); const membershipMeta = investmentResourceMeta({ resourceKey: 'accounts/membership', dataRevision: String(saved?.connection.membershipRevision ?? MISSING_MEMBERSHIP_REVISION), availability: available ? 'complete' : 'missing', fetchedAt: saved?.connection.membershipAt ?? null, freshForMs: tbankSourceTtlMs.membership, work: saved?.membershipWork ?? null, }); const meta = investmentResourceMeta({ resourceKey: `operations/${createHash('sha256').update(filterKey).digest('base64url')}`, dataRevision: saved?.binding ?? 'missing', availability: !available ? 'missing' : complete ? 'complete' : confirmed.length ? 'partial' : 'missing', fetchedAt: lastSuccessfulAt, freshForMs: plan.intervals.some((interval) => interval.stale) ? STALE_FRESHNESS_MS : Number.POSITIVE_INFINITY, work: work ?? saved?.membershipWork ?? null, 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, 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', }, }); const last = rows.at(LAST_ROW_INDEX); const hasNext = Boolean(saved && saved.rows.length > limit); return { data: meta.completeness === 'incomplete' ? null : { accountId, items: rows.map((row) => ({ ...mapBrokerageOperation( row.payload as TbankOperationItem, row.occurredAt, ), ...(accountId === 'all' ? { account: { id: row.accountId, name: accounts.find( (account) => account.accountId === row.accountId, )?.name ?? row.accountId, }, } : {}), logoName: instruments .get(metadataKey(row.payload as TbankOperationItem)) ?.data?.brand?.logoName?.trim() || null, })), hasNext, nextCursor: hasNext && last && saved ? this.encodeCursor({ binding: saved.binding, movingEnd, to: to.toISOString(), at: last.occurredAt.toISOString(), accountId: last.accountId, id: last.brokerOperationId, }) : null, observedAt: now.toISOString(), }, meta, }; } 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, ranges: (openedAt: Date | undefined) => DateOnlyRange[], observe = false, ) { const connection = await this.access.local(userId, accountId); const input = { userId, revision: connection.revision, accountId, from: new Date(UNIX_EPOCH_MS), to: new Date(), categories: ['income'], filterKey: `income/${accountId}`, periods: ranges, }; let saved = await this.repository.readPage(input); const schedule = async () => { const current = await this.access.local(userId, accountId); if (current.revision !== connection.revision) return; const snapshot = await this.repository.readPage(input); if (!snapshot?.connection.membershipAt) return; const targets = snapshot.connection.accounts.flatMap((account) => (snapshot.periods ?? []).map((period) => ({ accountId: account.accountId, ...period, })), ); if (targets.length) { await this.sync.readPlan(userId, targets, { movingEnd: snapshot.periods?.some((period) => period.movingEnd), history: true, revision: connection.revision, }); } }; if (!observe) { const discovery = await this.membership.request( userId, connection.revision, ); if (discovery) void discovery.done.then(schedule).catch(() => undefined); else await schedule(); saved = await this.repository.readPage(input); } const now = new Date(); const hourly = await this.hourlySync.hourly(userId); const membership = investmentResourceMeta({ resourceKey: 'accounts/membership', dataRevision: String(saved?.connection.membershipRevision ?? MISSING_MEMBERSHIP_REVISION), availability: saved?.connection.membershipAt ? 'complete' : 'missing', fetchedAt: saved?.connection.membershipAt ?? null, freshForMs: tbankSourceTtlMs.membership, work: saved?.membershipWork ?? null, }); // Plan even without a snapshot so cold responses retain the requested identity. const periods = saved?.periods ?? ranges(undefined).map((period) => ({ ...dateOnlyBounds(period), key: `${period.from}/${period.to}`, movingEnd: false, })); const rows = saved?.rows ?? []; const instruments = await this.metadata.savedMany( rows.map((row) => row.payload as TbankOperationItem), ); return periods.map((period) => { const targets = (saved?.connection.accounts ?? []).map((account) => ({ accountId: account.accountId, ...period, to: period.movingEnd ? now : period.to, })); const plan = planOperationRead(targets, saved?.coverage ?? [], now, { movingEnd: true, }); const relevant = (saved?.work ?? []).filter((work) => work.intervals.some((range) => targets.some((target) => operationIntervalsOverlap(range, target)), ), ); const work = relevant.find((work) => work.state === 'running') ?? relevant.find((work) => work.state === 'failed') ?? saved?.membershipWork ?? null; const known = Boolean(saved?.connection.membershipAt); const complete = known && plan.intervals.every((range) => range.availability === 'complete'); const sources = (saved?.coverage ?? []).filter( (row) => !row.instrumentUid && targets.some((target) => operationIntervalsOverlap(row, target)), ); const fetchedAt = sources .map((row) => row.fetchedAt) .sort((a, b) => a.valueOf() - b.valueOf())[0] ?? (complete ? saved?.connection.membershipAt : null) ?? null; const confirmed = confirmedOperationRanges(plan.intervals, sources); const meta = investmentResourceMeta({ resourceKey: `income/${accountId}/selected/${period.key}`, dataRevision: saved?.binding ?? 'missing', availability: !known ? 'missing' : complete ? 'complete' : sources.length ? 'partial' : 'missing', fetchedAt, freshForMs: plan.intervals.some((range) => range.stale) ? STALE_FRESHNESS_MS : Infinity, work, lastAttempt: hourlyAttempt(hourly, 'operations', accountId === 'all' ? undefined : accountId), dependencies: [membership], coverage: { requested: targets.map(serialize), completeRanges: confirmed.map(serialize), missingRanges: uncoveredOperationIntervals( plan.intervals, confirmed, ).map(serialize), asOf: complete && targets.length ? new Date( Math.min( ...targets.map((target) => Math.max( ...plan.intervals .filter( (range) => range.accountId === target.accountId, ) .map( (range) => range.asOf?.valueOf() ?? range.to.valueOf(), ), ), ), ), ).toISOString() : null, historyBoundary: (saved?.connection.accounts ?? []).every( (account) => account.openedOn, ) ? 'known' : 'unknown', historyState: complete ? 'complete' : sources.length ? 'partial' : 'missing', }, }); return { meta, items: complete ? rows .filter((row) => plan.intervals.some( (interval) => interval.accountId === row.accountId && row.occurredAt >= interval.from && row.occurredAt < interval.to, ), ) .map((row) => ({ ...mapBrokerageOperation( row.payload as TbankOperationItem, row.occurredAt, ), logoName: instruments .get(metadataKey(row.payload as TbankOperationItem)) ?.data?.brand?.logoName?.trim() || null, ...(accountId === 'all' ? { account: { id: row.accountId, name: saved?.connection.accounts.find( (account) => account.accountId === row.accountId, )?.name ?? row.accountId, }, } : {}), })) : null, }; }); } } function serialize(range: { accountId: string; from: Date; to: Date }) { return { accountId: range.accountId, from: range.from.toISOString(), to: range.to.toISOString() }; }