WealthLedger/backend
5
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: strengthen backend eslint profile

Browse Source
This commit is contained in:
ksv741
2026-09-28 11:04:18 +03:00
parent ca5dbcb382
commit 9b3558f96c
179 changed files with 4485 additions and 2716 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/investments/operations/application/brokerage-operations-sync.service.ts
+36 -27
View File
@@ -1,22 +1,23 @@
import { accountHistory } from '../../investment-sync-plan';
import { Injectable, NotFoundException } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
import {
TbankAccessService,
TbankClient,
} from '../../../../integrations/tbank/tbank.api';
import { accountHistory } from '../../investment-sync-plan';
import {
type OperationInterval,
operationIntervals,
planOperationRead,
uncoveredOperationIntervals,
} from '../domain/operations-sync-plan';
import {
BrokerageOperationsRepository,
type OperationAttempt,
} from '../infrastructure/brokerage-operations.repository';
import {
operationIntervals,
planOperationRead,
type OperationInterval,
uncoveredOperationIntervals,
} from '../domain/operations-sync-plan';
const operationsPageSize = 1000;
const priorities = { screen: 1, history: 2 } as const;
@@ -29,6 +30,7 @@ export class BrokerageOperationsSyncService {
string,
{ attempt: OperationAttempt; priority: number }
>();
private draining = false;
// Only short DB reservations/publications are serialized, never RPCs.
private reservations: Promise<unknown> = Promise.resolve();
@@ -44,10 +46,10 @@ export class BrokerageOperationsSyncService {
userId: string,
intervals: OperationInterval[],
options: {
movingEnd?: boolean;
force?: boolean;
observe?: boolean;
history?: boolean;
movingEnd?: boolean;
observe?: boolean;
revision?: string;
} = {},
) {
@@ -55,8 +57,9 @@ export class BrokerageOperationsSyncService {
if (
!connection.membershipAt ||
(options.revision && options.revision !== connection.revision)
)
) {
return { plan: null, handle: null };
}
if (
intervals.some(
(interval) =>
@@ -64,8 +67,9 @@ export class BrokerageOperationsSyncService {
(account) => account.accountId === interval.accountId,
),
)
)
) {
throw new NotFoundException();
}
const coverage = await this.repository.coverage(
userId,
connection.revision,
@@ -76,24 +80,24 @@ export class BrokerageOperationsSyncService {
...intervals.map((interval) => interval.to.valueOf()),
);
const targets = intervals.map((interval) =>
options.movingEnd && interval.to.valueOf() === movingTo
? { ...interval, to: now }
: interval,
options.movingEnd && interval.to.valueOf() === movingTo ?
{ ...interval, to: now } :
interval,
);
const plan = planOperationRead(targets, coverage, now, options);
const complete = plan.intervals.every(
(interval) => interval.availability === 'complete',
);
const handle =
!options.observe && plan.load.length
? await this.request(
!options.observe && plan.load.length ?
await this.request(
userId,
plan.load,
complete || Boolean(options.force),
'screen',
connection.revision,
)
: null;
) :
null;
if (!options.observe && !options.force && options.history) {
const accountIds = new Set(
intervals.map((interval) => interval.accountId),
@@ -108,7 +112,7 @@ export class BrokerageOperationsSyncService {
movingEnd: true,
}).load;
const remaining = uncoveredOperationIntervals(missing, targets);
if (remaining.length)
if (remaining.length) {
await this.request(
userId,
remaining,
@@ -116,6 +120,7 @@ export class BrokerageOperationsSyncService {
'history',
connection.revision,
);
}
}
return { plan, handle };
}
@@ -126,7 +131,7 @@ export class BrokerageOperationsSyncService {
atomic: boolean,
priority: keyof typeof priorities = 'screen',
revision?: string,
): Promise<string | null> {
): Promise<null | string> {
const connection = await this.access.local(userId);
if (
connection.membershipAt &&
@@ -136,8 +141,9 @@ export class BrokerageOperationsSyncService {
(account) => account.accountId === interval.accountId,
),
)
)
) {
throw new NotFoundException();
}
if (revision && revision !== connection.revision) return null;
const cutoff = new Date();
const bounded = operationIntervals(intervals)
@@ -155,6 +161,7 @@ export class BrokerageOperationsSyncService {
this.processRunId,
atomic,
this.runningAttempts,
// eslint-disable-next-line security/detect-object-injection -- The priority belongs to the internal priority union.
priorities[priority],
);
if (!requested) return null;
@@ -258,17 +265,18 @@ export class BrokerageOperationsSyncService {
Math.ceil(interval.to.valueOf() / MILLISECONDS_PER_SECOND) + 1,
),
},
...(interval.instrumentUid
? { instrumentId: interval.instrumentUid }
: {}),
...(interval.instrumentUid ?
{ instrumentId: interval.instrumentUid } :
{}),
cursor,
limit: operationsPageSize,
state: 'OPERATION_STATE_EXECUTED',
});
if (
!(await this.repository.stagePage(attempt, interval, cursor, response))
)
) {
return false;
}
if (!plan.atomic && !response.hasNext) {
if (
!(await this.repository.publish(
@@ -277,8 +285,9 @@ export class BrokerageOperationsSyncService {
new Date(),
true,
))
)
) {
return false;
}
}
// One page per turn: newly requested screen work can overtake backfill.
// TbankClient alone owns the bounded RPC queue and retries.
src/modules/investments/operations/application/brokerage-operations.service.ts
+136 -138
View File
@@ -1,41 +1,42 @@
import {
ledgerRange,
accountHistory,
planInvestmentScreen,
} from '../../investment-sync-plan';
import { ConfigService } from '@nestjs/config';
import { investmentResourceMeta } from '../../investment-resource.mapper';
import { BrokerageOperationsRepository } from '../infrastructure/brokerage-operations.repository';
import { BrokerageOperationsSyncService } from './brokerage-operations-sync.service';
import {
planOperationRead,
uncoveredOperationIntervals,
operationIntervalsOverlap,
} from '../domain/operations-sync-plan';
import { BadRequestException, Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { createHash, createHmac, timingSafeEqual } from 'node:crypto';
import {
InstrumentMetadataService,
metadataKey,
TbankAccessService,
TbankMembershipService,
tbankSourceTtlMs,
type TbankOperationItem,
tbankSourceTtlMs,
} from '../../../../integrations/tbank/tbank.api';
import { DateOnlyRange, dateOnlyBounds } from '../../investment-calendar';
import { dateOnlyBounds, DateOnlyRange } from '../../investment-calendar';
import { investmentResourceMeta } from '../../investment-resource.mapper';
import {
DEFAULT_OPERATIONS_PAGE_SIZE,
BrokerageOperationsQueryDto,
} from '../presentation/brokerage-operation.dto';
accountHistory,
ledgerRange,
planInvestmentScreen,
} from '../../investment-sync-plan';
import {
operationIntervalsOverlap,
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';
type LocalOperationsCursor = {
movingEnd?: boolean;
binding: string;
to: string;
at: string;
accountId: string;
at: string;
binding: string;
id: string;
movingEnd?: boolean;
to: string;
};
@Injectable()
@@ -70,18 +71,19 @@ export class BrokerageOperationsService {
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.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,
@@ -118,9 +120,9 @@ export class BrokerageOperationsService {
source.priority === 'screen' &&
source.accountId &&
source.from &&
source.to
? [{ accountId: source.accountId, from: source.from, to: source.to }]
: [],
source.to ?
[{ accountId: source.accountId, from: source.from, to: source.to }] :
[],
);
if (query.sync !== 'observe') {
const discovery = await this.membership.request(
@@ -148,9 +150,9 @@ export class BrokerageOperationsService {
accountId === 'all' || account.accountId === accountId,
)
.flatMap((account) =>
currentRange.from < currentRange.to
? [{ accountId: account.accountId, ...currentRange }]
: [],
currentRange.from < currentRange.to ?
[{ accountId: account.accountId, ...currentRange }] :
[],
);
await this.sync.readPlan(userId, targets, {
movingEnd,
@@ -170,16 +172,16 @@ export class BrokerageOperationsService {
const saved = await this.repository.readPage(readInput);
const accounts = saved?.connection.accounts ?? [];
const targets = accounts.flatMap((account) =>
from < to
? [
from < to ?
[
{
accountId: account.accountId,
from,
to,
instrumentUid: query.instrumentUid,
},
]
: [],
] :
[],
);
const plan = planOperationRead(targets, saved?.coverage ?? [], now, {
movingEnd,
@@ -191,8 +193,8 @@ export class BrokerageOperationsService {
row.accountId === target.accountId &&
(!row.instrumentUid ||
row.instrumentUid === target.instrumentUid) &&
row.from < target.to &&
target.from < row.to,
row.from < target.to &&
target.from < row.to,
)
.map((row) => ({
accountId: target.accountId,
@@ -230,13 +232,8 @@ export class BrokerageOperationsService {
sources
.map((row) => row.fetchedAt)
.sort((a, b) => a.valueOf() - b.valueOf())[0] ??
(complete ? saved?.connection.membershipAt : null) ??
null;
const serialize = (range: { accountId: string; from: Date; to: Date }) => ({
accountId: range.accountId,
from: range.from.toISOString(),
to: range.to.toISOString(),
});
(complete ? saved?.connection.membershipAt : null) ??
null;
const historyTargets = accountHistory(accounts, now);
const history = planOperationRead(
historyTargets,
@@ -273,9 +270,11 @@ export class BrokerageOperationsService {
] as const;
}),
);
for (const identity of identities.values())
if (!instruments.get(metadataKey(identity))?.fresh)
for (const identity of identities.values()) {
if (!instruments.get(metadataKey(identity))?.fresh) {
await this.metadata.request(userId, connection.revision, identity);
}
}
}
const membershipMeta = investmentResourceMeta({
resourceKey: 'accounts/membership',
@@ -288,17 +287,17 @@ export class BrokerageOperationsService {
const meta = investmentResourceMeta({
resourceKey: `operations/${createHash('sha256').update(filterKey).digest('base64url')}`,
dataRevision: saved?.binding ?? 'missing',
availability: !available
? 'missing'
: complete
? 'complete'
: confirmed.length
? 'partial'
: 'missing',
availability: !available ?
'missing' :
complete ?
'complete' :
confirmed.length ?
'partial' :
'missing',
fetchedAt: lastSuccessfulAt,
freshForMs: plan.intervals.some((interval) => interval.stale)
? 0
: Number.POSITIVE_INFINITY,
freshForMs: plan.intervals.some((interval) => interval.stale) ?
0 :
Number.POSITIVE_INFINITY,
work: work ?? saved?.membershipWork ?? null,
dependencies: [membershipMeta, ...metadataDependencies],
coverage: {
@@ -312,8 +311,8 @@ export class BrokerageOperationsService {
(interval) =>
interval.accountId === account.accountId && interval.asOf,
),
)
? new Date(
) ?
new Date(
Math.min(
...accounts.map((account) =>
Math.max(
@@ -326,42 +325,42 @@ export class BrokerageOperationsService {
),
),
),
).toISOString()
: null,
).toISOString() :
null,
historyBoundary: knownBoundary ? 'known' : 'unknown',
historyState: saved?.work.some(
(work) => work.state === 'running' && work.priority > 1,
)
? 'loading'
: saved?.work.some(
(work) => work.state === 'failed' && work.priority > 1,
)
? 'failed'
: knownBoundary &&
history.intervals.every(
(interval) => interval.availability === 'complete',
)
? 'complete'
: saved?.coverage.length
? 'partial'
: 'missing',
) ?
'loading' :
saved?.work.some(
(work) => work.state === 'failed' && work.priority > 1,
) ?
'failed' :
knownBoundary &&
history.intervals.every(
(interval) => interval.availability === 'complete',
) ?
'complete' :
saved?.coverage.length ?
'partial' :
'missing',
},
});
const last = rows.at(-1);
const hasNext = Boolean(saved && saved.rows.length > limit);
return {
data:
meta.availability === 'missing'
? null
: {
meta.availability === 'missing' ?
null :
{
accountId,
items: rows.map((row) => ({
...mapBrokerageOperation(
row.payload as TbankOperationItem,
row.occurredAt,
),
...(accountId === 'all'
? {
...(accountId === 'all' ?
{
account: {
id: row.accountId,
name:
@@ -369,8 +368,8 @@ export class BrokerageOperationsService {
(account) => account.accountId === row.accountId,
)?.name ?? row.accountId,
},
}
: {}),
} :
{}),
logoName:
instruments
.get(metadataKey(row.payload as TbankOperationItem))
@@ -378,16 +377,16 @@ export class BrokerageOperationsService {
})),
hasNext,
nextCursor:
hasNext && last && saved
? this.encodeCursor({
hasNext && last && saved ?
this.encodeCursor({
binding: saved.binding,
movingEnd,
to: to.toISOString(),
at: last.occurredAt.toISOString(),
accountId: last.accountId,
id: last.brokerOperationId,
})
: null,
}) :
null,
observedAt: now.toISOString(),
},
meta,
@@ -413,23 +412,25 @@ export class BrokerageOperationsService {
extra ||
actual.length !== expected.length ||
!timingSafeEqual(actual, expected)
)
) {
throw new Error();
}
const cursor = JSON.parse(
Buffer.from(body, 'base64url').toString(),
) as Partial<LocalOperationsCursor>;
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))
)
!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: 'INVALID_OPERATIONS_CURSOR' });
@@ -465,12 +466,13 @@ export class BrokerageOperationsService {
...period,
})),
);
if (targets.length)
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(
@@ -530,17 +532,8 @@ export class BrokerageOperationsService {
sources
.map((row) => row.fetchedAt)
.sort((a, b) => a.valueOf() - b.valueOf())[0] ??
(complete ? saved?.connection.membershipAt : null) ??
null;
const serialize = (range: {
accountId: string;
from: Date;
to: Date;
}) => ({
accountId: range.accountId,
from: range.from.toISOString(),
to: range.to.toISOString(),
});
(complete ? saved?.connection.membershipAt : null) ??
null;
const confirmed = plan.intervals.flatMap((target) =>
sources
.filter(
@@ -555,15 +548,16 @@ export class BrokerageOperationsService {
})),
);
const meta = investmentResourceMeta({
// eslint-disable-next-line security/detect-object-injection -- The numeric index enumerates the three comparison periods.
resourceKey: `income/${accountId}/${['selected', 'previous', 'yearAgo'][index]}/${period.key}`,
dataRevision: saved?.binding ?? 'missing',
availability: !known
? 'missing'
: complete
? 'complete'
: sources.length
? 'partial'
: 'missing',
availability: !known ?
'missing' :
complete ?
'complete' :
sources.length ?
'partial' :
'missing',
fetchedAt,
freshForMs: plan.intervals.some((range) => range.stale) ? 0 : Infinity,
work,
@@ -576,8 +570,8 @@ export class BrokerageOperationsService {
confirmed,
).map(serialize),
asOf:
complete && targets.length
? new Date(
complete && targets.length ?
new Date(
Math.min(
...targets.map((target) =>
Math.max(
@@ -592,24 +586,24 @@ export class BrokerageOperationsService {
),
),
),
).toISOString()
: null,
).toISOString() :
null,
historyBoundary: (saved?.connection.accounts ?? []).every(
(account) => account.openedOn,
)
? 'known'
: 'unknown',
historyState: complete
? 'complete'
: sources.length
? 'partial'
: 'missing',
) ?
'known' :
'unknown',
historyState: complete ?
'complete' :
sources.length ?
'partial' :
'missing',
},
});
return {
meta,
items: complete
? (saved?.rows ?? [])
items: complete ?
(saved?.rows ?? [])
.filter((row) =>
plan.intervals.some(
(interval) =>
@@ -623,8 +617,8 @@ export class BrokerageOperationsService {
row.payload as TbankOperationItem,
row.occurredAt,
),
...(accountId === 'all'
? {
...(accountId === 'all' ?
{
account: {
id: row.accountId,
name:
@@ -632,11 +626,15 @@ export class BrokerageOperationsService {
(account) => account.accountId === row.accountId,
)?.name ?? row.accountId,
},
}
: {}),
}))
: null,
} :
{}),
})) :
null,
};
});
}
}
function serialize(range: { accountId: string; from: Date; to: Date }) {
return { accountId: range.accountId, from: range.from.toISOString(), to: range.to.toISOString() };
}
src/modules/investments/operations/domain/operations-sync-plan.ts
+30 -25
View File
@@ -1,8 +1,8 @@
export type OperationInterval = {
accountId: string;
from: Date;
to: Date;
instrumentUid?: string;
to: Date;
};
// Normalize duplicate/overlapping targets before creating a durable identity.
@@ -12,8 +12,10 @@ export function operationIntervals(
): OperationInterval[] {
const groups = new Map<string, OperationInterval[]>();
for (const interval of requested) {
if (!(interval.from < interval.to))
// 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');
}
const key = JSON.stringify([
interval.accountId,
interval.instrumentUid ?? '',
@@ -23,17 +25,18 @@ export function operationIntervals(
groups.set(key, group);
}
const result: OperationInterval[] = [];
// 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.
for (const key of [...groups.keys()].sort()) {
const merged: OperationInterval[] = [];
for (const interval of (groups.get(key) ?? []).sort(
(left, right) => left.from.valueOf() - right.from.valueOf(),
)) {
const previous = merged.at(-1);
if (previous && interval.from <= previous.to)
if (previous && interval.from <= previous.to) {
previous.to = new Date(
Math.max(previous.to.valueOf(), interval.to.valueOf()),
);
else merged.push({ ...interval });
} else merged.push({ ...interval });
}
for (const interval of merged) {
let from = interval.from;
@@ -61,8 +64,8 @@ export function operationIntervalsOverlap(
(!left.instrumentUid ||
!right.instrumentUid ||
left.instrumentUid === right.instrumentUid) &&
left.from < right.to &&
right.from < left.to
left.from < right.to &&
right.from < left.to
);
}
@@ -78,15 +81,16 @@ export function uncoveredOperationIntervals(
!operationIntervalsOverlap(fragment, covered) ||
(covered.instrumentUid &&
covered.instrumentUid !== fragment.instrumentUid)
)
) {
return [fragment];
}
return [
...(fragment.from < covered.from
? [{ ...fragment, to: covered.from }]
: []),
...(covered.to < fragment.to
? [{ ...fragment, from: covered.to }]
: []),
...(fragment.from < covered.from ?
[{ ...fragment, to: covered.from }] :
[]),
...(covered.to < fragment.to ?
[{ ...fragment, from: covered.to }] :
[]),
];
});
}
@@ -95,8 +99,8 @@ export function uncoveredOperationIntervals(
}
export type OperationCoverage = OperationInterval & {
fetchedAt: Date;
asOf: Date;
fetchedAt: Date;
};
const operationsFreshForMs = 300_000;
@@ -104,7 +108,7 @@ export function planOperationRead(
requested: OperationInterval[],
coverage: OperationCoverage[],
now: Date,
options: { movingEnd?: boolean; force?: boolean } = {},
options: { force?: boolean; movingEnd?: boolean } = {},
) {
const monthStart = new Date(
Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1),
@@ -137,8 +141,9 @@ export function planOperationRead(
row.asOf.valueOf() === row.to.valueOf() && row.to > target.from,
)
.sort((left, right) => right.to.valueOf() - left.to.valueOf())[0];
if (latest && latest.to < target.to)
if (latest && latest.to < target.to) {
effective = { ...target, to: latest.to };
}
}
const missing = uncoveredOperationIntervals([effective], ranges);
const complete = missing.length === 0;
@@ -163,20 +168,20 @@ export function planOperationRead(
// Completeness is bounded by the observed end, never by the new wall clock.
return {
...effective,
availability: complete
? ('complete' as const)
: sources.length
? ('partial' as const)
: ('missing' as const),
availability: complete ?
('complete' as const) :
sources.length ?
('partial' as const) :
('missing' as const),
stale,
asOf: sources.length
? new Date(
asOf: sources.length ?
new Date(
Math.min(
effective.to.valueOf(),
Math.max(...sources.map((row) => row.asOf.valueOf())),
),
)
: null,
) :
null,
};
});
return { intervals, load: operationIntervals(load) };
src/modules/investments/operations/infrastructure/brokerage-operation.mapper.ts
+16 -16
View File
@@ -1,10 +1,10 @@
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
import { formatDecimal } from '../../../../common/decimal';
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
import { decimal } from '../../../../integrations/tbank/decimal';
import {
TbankMoney,
TbankOperationItem,
} from '../../../../integrations/tbank/tbank.api';
import { decimal } from '../../../../integrations/tbank/decimal';
import {
BrokerageOperationDto,
OperationCategory,
@@ -59,14 +59,14 @@ export function mapBrokerageOperation(
logoName: null,
payment: money(operation.payment),
quantity:
(operationCategory(operation.type) === 'trade'
? operation.quantityDone
: operation.quantity) === undefined
? null
: String(
operationCategory(operation.type) === 'trade'
? operation.quantityDone
: operation.quantity,
(operationCategory(operation.type) === 'trade' ?
operation.quantityDone :
operation.quantity) === undefined ?
null :
String(
operationCategory(operation.type) === 'trade' ?
operation.quantityDone :
operation.quantity,
),
price: money(operation.price),
accruedInterest: money(operation.accruedInt),
@@ -74,22 +74,22 @@ export function mapBrokerageOperation(
};
}
function text(value: string | undefined): string | null {
function text(value: string | undefined): null | string {
return value?.trim() || null;
}
function money(value: TbankMoney | undefined) {
const currency = value?.currency?.trim().toUpperCase();
return currency && value
? { currency, value: formatDecimal(decimal(value)) }
: null;
return currency && value ?
{ currency, value: formatDecimal(decimal(value)) } :
null;
}
export function operationTimestamp(
value: { seconds?: string | number; nanos?: number } | undefined,
value: undefined | { nanos?: number; seconds?: number | string },
) {
return new Date(
Number(value?.seconds ?? 0) * MILLISECONDS_PER_SECOND +
Math.floor((value?.nanos ?? 0) / NANOSECONDS_PER_MILLISECOND),
Math.floor((value?.nanos ?? 0) / NANOSECONDS_PER_MILLISECOND),
);
}
src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts
+142 -114
View File
@@ -5,46 +5,48 @@ import {
} from '@nestjs/common';
import { createHash, randomUUID } from 'node:crypto';
import { isDeepStrictEqual } from 'node:util';
import { PrismaService } from '../../../../prisma/prisma.service';
import type { Prisma } from '../../../../../generated/prisma/client';
import type {
TbankOperationItem,
TbankOperationsByCursorResponse,
} from '../../../../integrations/tbank/tbank.api';
import { PrismaService } from '../../../../prisma/prisma.service';
import {
DateOnlyRange,
dateOnlyBounds,
DateOnlyRange,
moscowDate,
} from '../../investment-calendar';
import {
operationCategory,
operationTimestamp,
} from './brokerage-operation.mapper';
import {
type OperationInterval,
operationIntervals,
operationIntervalsOverlap,
uncoveredOperationIntervals,
} from '../domain/operations-sync-plan';
import {
operationCategory,
operationTimestamp,
} from './brokerage-operation.mapper';
export type { OperationInterval } from '../domain/operations-sync-plan';
export type OperationAttempt = {
id: string;
attemptId: string;
userId: string;
id: string;
revision: string;
userId: string;
};
type OperationsIntent = {
intervals: string[];
atomic: boolean;
atomicIntervals?: string[];
intervals: string[];
screenIntervals?: string[];
};
type PageProgress = {
complete: boolean;
cursor: string;
seen: string[];
pages: number;
seen: string[];
};
const progressItem = '';
@@ -53,17 +55,17 @@ export class BrokerageOperationsRepository {
constructor(private readonly prisma: PrismaService) {}
async readPage(input: {
userId: string;
revision: string;
accountId: string;
from: Date;
to: Date;
instrumentUid?: 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[];
filterKey: string;
cursor?: { binding: string; at: string; accountId: string; id: string };
revision: string;
to: Date;
userId: string;
}) {
return this.prisma.$transaction(async (tx) => {
const connection = await tx.tbankConnection.findUnique({
@@ -92,14 +94,16 @@ export class BrokerageOperationsRepository {
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;
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);
@@ -123,8 +127,9 @@ export class BrokerageOperationsRepository {
]),
)
.digest('base64url');
if (input.cursor && input.cursor.binding !== binding)
if (input.cursor && input.cursor.binding !== binding) {
throw new BadRequestException({ code: 'INVALID_OPERATIONS_CURSOR' });
}
const scope = {
userId: input.userId,
revision: input.revision,
@@ -162,21 +167,21 @@ export class BrokerageOperationsRepository {
const rows = await tx.tbankOperation.findMany({
where: {
...scope,
...(periods
? {
...(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
? {
} :
{ 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) } },
{
@@ -189,8 +194,8 @@ export class BrokerageOperationsRepository {
brokerOperationId: { gt: after.id },
},
],
}
: {}),
} :
{}),
},
orderBy: [
{ occurredAt: 'desc' },
@@ -247,6 +252,7 @@ export class BrokerageOperationsRepository {
const scopeKey =
'operations:' +
createHash('sha256')
// 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.
.update(JSON.stringify([...new Set(intervals.map(intervalKey))].sort()))
.digest('hex');
return this.prisma.$transaction(async (tx) => {
@@ -271,14 +277,15 @@ export class BrokerageOperationsRepository {
(account) => account.accountId === interval.accountId,
),
)
)
) {
throw new NotFoundException();
}
if (priority > 1) {
for (const account of accounts.filter((account) =>
intervals.some(
(interval) => interval.accountId === account.accountId,
),
))
)) {
await tx.tbankAccount.update({
where: {
userId_revision_accountId: {
@@ -289,13 +296,14 @@ export class BrokerageOperationsRepository {
},
data: {
historyFrom: new Date(
account.openedOn
? `${account.openedOn}T00:00:00Z`
: '1970-01-01T00:00:00Z',
account.openedOn ?
`${account.openedOn}T00:00:00Z` :
'1970-01-01T00:00:00Z',
),
historyBoundary: account.openedOn ? 'known' : 'unknown',
},
});
}
}
const coverage = await tx.tbankOperationCoverage.findMany({
where: {
@@ -366,42 +374,42 @@ export class BrokerageOperationsRepository {
) {
const pending = previousIntent?.intervals.map(decodeInterval) ?? [];
const additions =
previous.scopeKey === scopeKey
? []
: uncoveredOperationIntervals(intervals, pending);
previous.scopeKey === scopeKey ?
[] :
uncoveredOperationIntervals(intervals, pending);
const work = await tx.tbankSyncWork.update({
where: { id: previous.id },
data: {
intent: {
atomic: previousIntent?.atomic ?? publicationAtomic,
atomicIntervals: previousIntent?.atomic
? [
atomicIntervals: previousIntent?.atomic ?
[
...new Set([
...(previousIntent.atomicIntervals ??
previousIntent.intervals),
...additions.map(intervalKey),
]),
]
: [],
] :
[],
screenIntervals: [
...new Set([
...(previousIntent?.screenIntervals ??
(previous.priority === 1
? (previousIntent?.intervals ?? [])
: [])),
...(priority === 1
? [...pending, ...additions]
(previous.priority === 1 ?
(previousIntent?.intervals ?? []) :
[])),
...(priority === 1 ?
[...pending, ...additions]
.filter((item) =>
intervals.some((target) =>
operationIntervalsOverlap(item, target),
),
)
.map(intervalKey)
: []),
.map(intervalKey) :
[]),
]),
],
intervals: (priority < previous.priority
? [
intervals: (priority < previous.priority ?
[
...pending.filter((item) =>
intervals.some((target) =>
operationIntervalsOverlap(item, target),
@@ -414,8 +422,8 @@ export class BrokerageOperationsRepository {
operationIntervalsOverlap(item, target),
),
),
]
: [...pending, ...additions]
] :
[...pending, ...additions]
).map(intervalKey),
},
priority: Math.min(priority, previous.priority),
@@ -427,9 +435,9 @@ export class BrokerageOperationsRepository {
previous &&
previous.state !== 'complete' &&
previous.state !== 'superseded';
const retained = recovering
? (previousIntent?.intervals.map(decodeInterval) ?? [])
: [];
const retained = recovering ?
(previousIntent?.intervals.map(decodeInterval) ?? []) :
[];
// A bridge between two active groups must fence both producers before
// either can publish a subset of the newly combined result.
for (const joined of touching) {
@@ -448,26 +456,27 @@ export class BrokerageOperationsRepository {
});
await tx.tbankSyncStage.deleteMany({ where: { workId: joined.id } });
}
if (previous)
if (previous) {
await tx.tbankSyncStage.deleteMany({ where: { workId: previous.id } });
}
const targets =
recovering && !upgradeGroup && touching.length <= 1
? previous?.scopeKey === scopeKey
? retained
: [...retained, ...uncoveredOperationIntervals(intervals, retained)]
: operationIntervals([...retained, ...intervals]);
recovering && !upgradeGroup && touching.length <= 1 ?
previous?.scopeKey === scopeKey ?
retained :
[...retained, ...uncoveredOperationIntervals(intervals, retained)] :
operationIntervals([...retained, ...intervals]);
const atomicTargets = [
...(publicationAtomic ? intervals : []),
...(recovering && previousIntent?.atomic
? (previousIntent.atomicIntervals ?? previousIntent.intervals).map(
...(recovering && previousIntent?.atomic ?
(previousIntent.atomicIntervals ?? previousIntent.intervals).map(
decodeInterval,
)
: []),
) :
[]),
...touching.flatMap((work) => {
const joined = work.intent as OperationsIntent;
return joined.atomic
? (joined.atomicIntervals ?? joined.intervals).map(decodeInterval)
: [];
return joined.atomic ?
(joined.atomicIntervals ?? joined.intervals).map(decodeInterval) :
[];
}),
];
const intent = {
@@ -486,14 +495,14 @@ export class BrokerageOperationsRepository {
intervals.some((incoming) =>
operationIntervalsOverlap(target, incoming),
)) ||
(
previousIntent?.screenIntervals ??
(previous?.priority === 1
? (previousIntent?.intervals ?? [])
: [])
).some((key) =>
operationIntervalsOverlap(target, decodeInterval(key)),
),
(
previousIntent?.screenIntervals ??
(previous?.priority === 1 ?
(previousIntent?.intervals ?? []) :
[])
).some((key) =>
operationIntervalsOverlap(target, decodeInterval(key)),
),
)
.map(intervalKey),
atomic: atomicTargets.length > 0,
@@ -545,20 +554,20 @@ export class BrokerageOperationsRepository {
marker.payload as PageProgress,
]),
);
const group = intent.atomic
? (intent.atomicIntervals ?? intent.intervals)
: intent.intervals;
const group = intent.atomic ?
(intent.atomicIntervals ?? intent.intervals) :
intent.intervals;
const nextKey = group.find((key) => !progress.get(key)?.complete);
return {
priority: work.priority,
atomic: intent.atomic,
intervals: group.map(decodeInterval),
next: nextKey
? {
next: nextKey ?
{
interval: decodeInterval(nextKey),
cursor: progress.get(nextKey)?.cursor || undefined,
}
: null,
} :
null,
};
}
@@ -582,8 +591,9 @@ export class BrokerageOperationsRepository {
updatedAt: true,
},
});
if (!work || work.state !== 'superseded')
if (!work || work.state !== 'superseded') {
return work ? { ...work, id } : null;
}
target = (work.progress as { successorId: string }).successorId;
}
throw new Error('Cyclic operations work successor');
@@ -599,7 +609,7 @@ export class BrokerageOperationsRepository {
errorCode: invalidToken ? 'INVALID_TBANK_TOKEN' : 'TBANK_UNAVAILABLE',
},
});
if (invalidToken)
if (invalidToken) {
await tx.tbankConnection.update({
where: {
userId_revision: {
@@ -609,6 +619,7 @@ export class BrokerageOperationsRepository {
},
data: { invalidToken: true },
});
}
return true;
});
}
@@ -636,34 +647,40 @@ export class BrokerageOperationsRepository {
},
});
const progress = marker?.payload as PageProgress | undefined;
if (progress?.complete || (progress?.cursor ?? '') !== (cursor ?? ''))
if (progress?.complete || (progress?.cursor ?? '') !== (cursor ?? '')) {
throw new Error('Unexpected operations page');
if (response.hasNext === undefined)
}
if (response.hasNext === undefined) {
throw new Error('Missing operations pagination state');
}
const nextCursor = response.nextCursor ?? '';
if (
response.hasNext &&
(!nextCursor ||
nextCursor === cursor ||
progress?.seen.includes(nextCursor))
)
) {
throw new Error('Invalid operations continuation');
}
for (const operation of response.items ?? []) {
if (!operation.id?.trim())
if (!operation.id?.trim()) {
throw new Error('Missing broker operation identity');
}
const occurredAt = operationTimestamp(operation.date);
if (
operation.date?.seconds === undefined ||
!Number.isFinite(occurredAt.valueOf())
)
) {
throw new Error('Invalid operation timestamp');
}
if (
occurredAt < interval.from ||
occurredAt >= interval.to ||
(interval.instrumentUid &&
operation.instrumentUid !== interval.instrumentUid)
)
) {
continue;
}
const payload = JSON.parse(
JSON.stringify({ ...operation, cursor: undefined }),
) as Prisma.InputJsonObject;
@@ -672,10 +689,12 @@ export class BrokerageOperationsRepository {
const previous = await tx.tbankSyncStage.findUnique({
where: { workId_attemptId_sourceKey_itemKey: identity },
});
if (previous && !isDeepStrictEqual(previous.payload, payload))
if (previous && !isDeepStrictEqual(previous.payload, payload)) {
throw new Error('Conflicting broker operation identity');
if (!previous)
}
if (!previous) {
await tx.tbankSyncStage.create({ data: { ...identity, payload } });
}
}
const payload: PageProgress = {
complete: !response.hasNext,
@@ -712,8 +731,9 @@ export class BrokerageOperationsRepository {
fetchedAt: Date,
advanceWork = false,
): Promise<boolean> {
if (!intervals.length)
if (!intervals.length) {
throw new Error('Empty operations publication group');
}
return this.prisma.$transaction(
async (tx) => {
if (!(await claim(tx, attempt))) return false;
@@ -734,8 +754,9 @@ export class BrokerageOperationsRepository {
},
},
});
if (!(marker?.payload as PageProgress | undefined)?.complete)
if (!(marker?.payload as PageProgress | undefined)?.complete) {
throw new Error('Incomplete operations publication group');
}
const staged = await tx.tbankSyncStage.findMany({
where: { ...stage, itemKey: { not: progressItem } },
});
@@ -745,10 +766,11 @@ export class BrokerageOperationsRepository {
if (
previous !== undefined &&
!isDeepStrictEqual(previous, row.payload)
)
) {
throw new Error(
'Conflicting broker operation identity across intervals',
);
}
groupRows.set(identity, row.payload);
}
const rows = staged.filter((row) => {
@@ -764,16 +786,16 @@ export class BrokerageOperationsRepository {
where: {
...scope,
occurredAt: { gte: interval.from, lt: interval.to },
...(interval.instrumentUid
? { instrumentUid: interval.instrumentUid }
: {}),
...(interval.instrumentUid ?
{ instrumentUid: interval.instrumentUid } :
{}),
},
});
const incoming = new Set(rows.map((row) => row.itemKey));
const removed = existing.filter(
(row) => !incoming.has(row.brokerOperationId),
);
for (const row of removed)
for (const row of removed) {
await tx.tbankOperation.delete({
where: {
userId_revision_accountId_brokerOperationId: {
@@ -782,6 +804,7 @@ export class BrokerageOperationsRepository {
},
},
});
}
if (removed.length) changedAccounts.add(interval.accountId);
for (const row of rows) {
const operation = row.payload as TbankOperationItem;
@@ -789,8 +812,9 @@ export class BrokerageOperationsRepository {
const previous = await tx.tbankOperation.findUnique({
where: { userId_revision_accountId_brokerOperationId: identity },
});
if (previous && isDeepStrictEqual(previous.payload, operation))
if (previous && isDeepStrictEqual(previous.payload, operation)) {
continue;
}
const data = {
occurredAt: operationTimestamp(operation.date),
instrumentUid: operation.instrumentUid ?? null,
@@ -818,7 +842,7 @@ export class BrokerageOperationsRepository {
update: { fetchedAt, asOf: interval.to },
});
}
for (const accountId of changedAccounts)
for (const accountId of changedAccounts) {
await tx.tbankAccount.update({
where: {
userId_revision_accountId: {
@@ -829,7 +853,8 @@ export class BrokerageOperationsRepository {
},
data: { operationsRevision: { increment: 1 } },
});
for (const interval of intervals)
}
for (const interval of intervals) {
await tx.tbankSyncStage.deleteMany({
where: {
workId: attempt.id,
@@ -837,6 +862,7 @@ export class BrokerageOperationsRepository {
sourceKey: intervalKey(interval),
},
});
}
if (advanceWork) {
const work = await tx.tbankSyncWork.findUniqueOrThrow({
where: { id: attempt.id },
@@ -867,9 +893,9 @@ export class BrokerageOperationsRepository {
intent.screenIntervals ??
(work.priority === 1 ? intent.intervals : [])
).includes(key),
)
? 1
: 2,
) ?
1 :
2,
progress: { remainingIntervals: remaining.length },
},
});
@@ -882,8 +908,10 @@ export class BrokerageOperationsRepository {
}
function intervalKey(interval: OperationInterval): string {
if (!(interval.from < interval.to))
// 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(),
src/modules/investments/operations/operations.api.ts
+3 -3
View File
@@ -1,8 +1,8 @@
// Public cross-module surface. Nest providers are exported by the matching module.
export { BrokerageOperationDto } from './presentation/brokerage-operation.dto';
export { BrokerageOperationsSyncService } from './application/brokerage-operations-sync.service';
export { BrokerageOperationsService } from './application/brokerage-operations.service';
export type { OperationInterval } from './domain/operations-sync-plan';
// Public cross-module surface. Nest providers are exported by the matching module.
export { BrokerageOperationDto } from './presentation/brokerage-operation.dto';
export { operationCategories } from './presentation/brokerage-operation.dto';
export { BrokerageOperationsSyncService } from './application/brokerage-operations-sync.service';
src/modules/investments/operations/operations.module.ts
+3 -2
View File
@@ -1,10 +1,11 @@
import { Module } from '@nestjs/common';
import { TbankModule } from '../../../integrations/tbank/tbank.module';
import { AuthModule } from '../../auth/auth.module';
import { BrokerageOperationsController } from './presentation/brokerage-operations.controller';
import { BrokerageOperationsSyncService } from './application/brokerage-operations-sync.service';
import { BrokerageOperationsService } from './application/brokerage-operations.service';
import { BrokerageOperationsRepository } from './infrastructure/brokerage-operations.repository';
import { BrokerageOperationsSyncService } from './application/brokerage-operations-sync.service';
import { BrokerageOperationsController } from './presentation/brokerage-operations.controller';
@Module({
imports: [AuthModule, TbankModule],
src/modules/investments/operations/presentation/brokerage-operation.dto.ts
+13 -12
View File
@@ -1,7 +1,3 @@
import {
InvestmentReadQueryDto,
InvestmentResourceMetaDto,
} from '../../investment-resource.dto';
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
import { Transform, Type } from 'class-transformer';
import {
@@ -16,9 +12,13 @@ import {
MaxLength,
Min,
} from 'class-validator';
import { MoneyDto } from '../../../../common/money.dto';
import { MoneyDto } from '../../../../common/money.dto';
import { AccountReferenceDto } from '../../account-reference.dto';
import {
InvestmentReadQueryDto,
InvestmentResourceMetaDto,
} from '../../investment-resource.dto';
const MAX_CURSOR_LENGTH = 4096;
const MAX_OPERATIONS_PAGE_SIZE = 1000;
@@ -102,25 +102,25 @@ export class BrokerageOperationDto {
declare description: string;
@ApiProperty({ type: String, nullable: true })
declare figi: string | null;
declare figi: null | string;
@ApiProperty({ type: String, nullable: true })
declare instrumentUid: string | null;
declare instrumentUid: null | string;
@ApiProperty({ type: String, nullable: true })
declare ticker: string | null;
declare ticker: null | string;
@ApiProperty({ type: String, nullable: true })
declare instrumentType: string | null;
declare instrumentType: null | string;
@ApiProperty({ type: String, nullable: true })
declare logoName: string | null;
declare logoName: null | string;
@ApiProperty({ type: MoneyDto, nullable: true })
declare payment: MoneyDto | null;
@ApiProperty({ type: String, nullable: true })
declare quantity: string | null;
declare quantity: null | string;
@ApiProperty({ type: MoneyDto, nullable: true })
declare price: MoneyDto | null;
@@ -140,7 +140,7 @@ export class BrokerageOperationsPageDto {
declare items: BrokerageOperationDto[];
@ApiProperty({ type: String, nullable: true })
declare nextCursor: string | null;
declare nextCursor: null | string;
@ApiProperty()
declare hasNext: boolean;
@@ -152,6 +152,7 @@ export class BrokerageOperationsPageDto {
export class BrokerageOperationsResourceDto {
@ApiProperty({ type: BrokerageOperationsPageDto, nullable: true })
declare data: BrokerageOperationsPageDto | null;
@ApiProperty({ type: InvestmentResourceMetaDto })
declare meta: InvestmentResourceMetaDto;
}
src/modules/investments/operations/presentation/brokerage-operations.controller.ts
+7 -5
View File
@@ -1,13 +1,15 @@
import type { Request } from 'express';
import { Controller, Get, Param, Query, Req, UseGuards } from '@nestjs/common';
import { ApiBearerAuth, ApiOkResponse, ApiTags } from '@nestjs/swagger';
import type { Request } from 'express';
import { ApiProblemResponses } from '../../../../common/decorators/api-safe-error-response.decorator';
import { AccessPayload, JwtAuthGuard } from '../../../auth/auth.api';
import {
BrokerageOperationsResourceDto,
BrokerageOperationsQueryDto,
} from './brokerage-operation.dto';
import { BrokerageOperationsService } from '../application/brokerage-operations.service';
import {
BrokerageOperationsQueryDto,
BrokerageOperationsResourceDto,
} from './brokerage-operation.dto';
@ApiTags('brokerage-accounts')
@ApiProblemResponses()