956 lines
31 KiB
TypeScript
956 lines
31 KiB
TypeScript
import {
|
|
BadRequestException,
|
|
Injectable,
|
|
NotFoundException,
|
|
} from '@nestjs/common';
|
|
import { createHash, randomUUID } from 'node:crypto';
|
|
import { isDeepStrictEqual } from 'node:util';
|
|
|
|
import type { Prisma } from '../../../../../generated/prisma/client';
|
|
import type {
|
|
TbankOperationItem,
|
|
TbankOperationsByCursorResponse,
|
|
} from '../../../../integrations/tbank/tbank.api';
|
|
|
|
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';
|
|
import {
|
|
type OperationInterval,
|
|
operationIntervals,
|
|
operationIntervalsOverlap,
|
|
operationsSyncPriority,
|
|
uncoveredOperationIntervals,
|
|
} from '../domain/operations-sync-plan';
|
|
import {
|
|
operationCategory,
|
|
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 { OperationInterval } from '../domain/operations-sync-plan';
|
|
|
|
export type OperationAttempt = {
|
|
attemptId: string;
|
|
id: string;
|
|
revision: string;
|
|
userId: string;
|
|
};
|
|
type OperationsIntent = {
|
|
atomic: boolean;
|
|
atomicIntervals?: string[];
|
|
intervals: string[];
|
|
screenIntervals?: string[];
|
|
};
|
|
type PageProgress = {
|
|
complete: boolean;
|
|
cursor: string;
|
|
pages: number;
|
|
seen: string[];
|
|
};
|
|
const progressItem = '';
|
|
|
|
@Injectable()
|
|
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: {
|
|
userId,
|
|
revision,
|
|
accountId: { in: accountIds },
|
|
account: { connection: { invalidToken: false } },
|
|
},
|
|
});
|
|
return rows.map(({ instrumentPredicate, ...row }) => ({
|
|
...row,
|
|
instrumentUid: instrumentPredicate || undefined,
|
|
}));
|
|
}
|
|
|
|
async request(
|
|
userId: string,
|
|
revision: string,
|
|
intervals: OperationInterval[],
|
|
processRunId: string,
|
|
atomic: boolean,
|
|
runningAttempts: ReadonlySet<string>,
|
|
priority: number,
|
|
) {
|
|
if (!intervals.length) throw new Error('Empty operations request');
|
|
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) => {
|
|
const locked = await tx.tbankConnection.updateMany({
|
|
where: {
|
|
userId,
|
|
revision,
|
|
invalidToken: false,
|
|
membershipAt: { not: null },
|
|
},
|
|
data: { membershipRevision: { increment: 0 } },
|
|
});
|
|
if (!locked.count) return null;
|
|
const accounts = await tx.tbankAccount.findMany({
|
|
where: { userId, revision },
|
|
select: { accountId: true, openedOn: true },
|
|
});
|
|
if (
|
|
intervals.some(
|
|
(interval) =>
|
|
!accounts.some(
|
|
(account) => account.accountId === interval.accountId,
|
|
),
|
|
)
|
|
) {
|
|
throw new NotFoundException();
|
|
}
|
|
if (priority > operationsSyncPriority.screen) {
|
|
for (const account of accounts.filter((account) =>
|
|
intervals.some(
|
|
(interval) => interval.accountId === account.accountId,
|
|
),
|
|
)) {
|
|
await tx.tbankAccount.update({
|
|
where: {
|
|
userId_revision_accountId: {
|
|
userId,
|
|
revision,
|
|
accountId: account.accountId,
|
|
},
|
|
},
|
|
data: {
|
|
historyFrom: new Date(
|
|
account.openedOn
|
|
? `${account.openedOn}T00:00:00Z`
|
|
: '1970-01-01T00:00:00Z',
|
|
),
|
|
historyBoundary: account.openedOn ? 'known' : 'unknown',
|
|
},
|
|
});
|
|
}
|
|
}
|
|
const coverage = await tx.tbankOperationCoverage.findMany({
|
|
where: {
|
|
userId,
|
|
revision,
|
|
accountId: { in: intervals.map((interval) => interval.accountId) },
|
|
},
|
|
});
|
|
const publicationAtomic =
|
|
atomic ||
|
|
intervals.every(
|
|
(interval) =>
|
|
uncoveredOperationIntervals(
|
|
[interval],
|
|
coverage
|
|
.filter(
|
|
(row) =>
|
|
row.accountId === interval.accountId &&
|
|
(!row.instrumentPredicate ||
|
|
row.instrumentPredicate === interval.instrumentUid),
|
|
)
|
|
.map((row) => ({
|
|
accountId: row.accountId,
|
|
from: row.from,
|
|
to: row.to,
|
|
instrumentUid: interval.instrumentUid,
|
|
})),
|
|
).length === EMPTY_INTERVAL_COUNT,
|
|
);
|
|
const active = await tx.tbankSyncWork.findMany({
|
|
where: {
|
|
userId,
|
|
revision,
|
|
state: 'running',
|
|
scopeKey: { startsWith: 'operations:' },
|
|
},
|
|
orderBy: [{ createdAt: 'asc' }, { id: 'asc' }],
|
|
});
|
|
const touching = active.filter((work) => {
|
|
const pending = (work.intent as { intervals: string[] }).intervals.map(
|
|
decodeInterval,
|
|
);
|
|
return pending.some((existing) =>
|
|
intervals.some((incoming) =>
|
|
operationIntervalsOverlap(existing, incoming),
|
|
),
|
|
);
|
|
});
|
|
const previous =
|
|
touching[0] ??
|
|
(await tx.tbankSyncWork.findUnique({
|
|
where: { userId_revision_scopeKey: { userId, revision, scopeKey } },
|
|
}));
|
|
const identity = {
|
|
userId,
|
|
revision,
|
|
scopeKey: previous?.scopeKey ?? scopeKey,
|
|
};
|
|
const previousIntent = previous?.intent as OperationsIntent | undefined;
|
|
const upgradeGroup =
|
|
publicationAtomic && previousIntent && !previousIntent.atomic;
|
|
if (
|
|
touching.length <= SINGLE_WORK_GROUP &&
|
|
previous?.state === 'running' &&
|
|
previous.processRunId === processRunId &&
|
|
runningAttempts.has(previous.attemptId) &&
|
|
!upgradeGroup
|
|
) {
|
|
const pending = previousIntent?.intervals.map(decodeInterval) ?? [];
|
|
const additions =
|
|
previous.scopeKey === scopeKey
|
|
? []
|
|
: uncoveredOperationIntervals(intervals, pending);
|
|
const work = await tx.tbankSyncWork.update({
|
|
where: { id: previous.id },
|
|
data: {
|
|
intent: extendRunningIntent({
|
|
previous: previousIntent,
|
|
publicationAtomic,
|
|
pending,
|
|
additions,
|
|
intervals,
|
|
priority,
|
|
previousPriority: previous.priority,
|
|
}),
|
|
priority: Math.min(priority, previous.priority),
|
|
},
|
|
});
|
|
return { work, launch: false };
|
|
}
|
|
const recovering =
|
|
previous &&
|
|
previous.state !== 'complete' &&
|
|
previous.state !== 'superseded';
|
|
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) {
|
|
if (joined.id === previous?.id) continue;
|
|
retained.push(
|
|
...(joined.intent as { intervals: string[] }).intervals.map(
|
|
decodeInterval,
|
|
),
|
|
);
|
|
await tx.tbankSyncWork.update({
|
|
where: { id: joined.id },
|
|
data: {
|
|
state: 'superseded',
|
|
progress: { successorId: previous?.id ?? '' },
|
|
},
|
|
});
|
|
await tx.tbankSyncStage.deleteMany({ where: { workId: joined.id } });
|
|
}
|
|
if (previous) {
|
|
await tx.tbankSyncStage.deleteMany({ where: { workId: previous.id } });
|
|
}
|
|
const targets =
|
|
recovering && !upgradeGroup && touching.length <= SINGLE_WORK_GROUP
|
|
? previous?.scopeKey === scopeKey
|
|
? retained
|
|
: [...retained, ...uncoveredOperationIntervals(intervals, retained)]
|
|
: operationIntervals([...retained, ...intervals]);
|
|
const atomicTargets = [
|
|
...(publicationAtomic ? intervals : []),
|
|
...(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)
|
|
: [];
|
|
}),
|
|
];
|
|
const intent = {
|
|
intervals: targets.map(intervalKey),
|
|
atomicIntervals: targets
|
|
.filter((target) =>
|
|
atomicTargets.some((atomic) =>
|
|
operationIntervalsOverlap(target, atomic),
|
|
),
|
|
)
|
|
.map(intervalKey),
|
|
screenIntervals: targets
|
|
.filter(
|
|
(target) =>
|
|
(priority === operationsSyncPriority.screen &&
|
|
intervals.some((incoming) =>
|
|
operationIntervalsOverlap(target, incoming),
|
|
)) ||
|
|
(
|
|
previousIntent?.screenIntervals ??
|
|
(previous?.priority === operationsSyncPriority.screen
|
|
? (previousIntent?.intervals ?? [])
|
|
: [])
|
|
).some((key) =>
|
|
operationIntervalsOverlap(target, decodeInterval(key)),
|
|
),
|
|
)
|
|
.map(intervalKey),
|
|
atomic: atomicTargets.length > EMPTY_INTERVAL_COUNT,
|
|
};
|
|
const data = {
|
|
intent,
|
|
state: 'running',
|
|
attemptId: randomUUID(),
|
|
processRunId,
|
|
priority: Math.min(
|
|
priority,
|
|
previous?.priority ?? priority,
|
|
...touching.map((work) => work.priority),
|
|
),
|
|
progress: {},
|
|
errorCode: null,
|
|
};
|
|
const work = await tx.tbankSyncWork.upsert({
|
|
where: { userId_revision_scopeKey: identity },
|
|
create: { ...identity, ...data, id: randomUUID() },
|
|
update: data,
|
|
});
|
|
return { work, launch: true };
|
|
});
|
|
}
|
|
|
|
async pending(attempt: OperationAttempt) {
|
|
const work = await this.prisma.tbankSyncWork.findFirst({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: false },
|
|
},
|
|
select: { intent: true, priority: true },
|
|
});
|
|
if (!work) return null;
|
|
const intent = work.intent as OperationsIntent;
|
|
const markers = await this.prisma.tbankSyncStage.findMany({
|
|
where: {
|
|
workId: attempt.id,
|
|
attemptId: attempt.attemptId,
|
|
itemKey: progressItem,
|
|
},
|
|
select: { sourceKey: true, payload: true },
|
|
});
|
|
const progress = new Map(
|
|
markers.map((marker) => [
|
|
marker.sourceKey,
|
|
marker.payload as PageProgress,
|
|
]),
|
|
);
|
|
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
|
|
? {
|
|
interval: decodeInterval(nextKey),
|
|
cursor: progress.get(nextKey)?.cursor || undefined,
|
|
}
|
|
: null,
|
|
};
|
|
}
|
|
|
|
async observe(userId: string, revision: string, id: string) {
|
|
let target = id;
|
|
const seen = new Set<string>();
|
|
while (!seen.has(target)) {
|
|
seen.add(target);
|
|
const work = await this.prisma.tbankSyncWork.findFirst({
|
|
where: {
|
|
id: target,
|
|
userId,
|
|
revision,
|
|
scopeKey: { startsWith: 'operations:' },
|
|
},
|
|
select: {
|
|
id: true,
|
|
state: true,
|
|
progress: true,
|
|
errorCode: true,
|
|
updatedAt: true,
|
|
},
|
|
});
|
|
if (!work || work.state !== 'superseded') {
|
|
return work ? { ...work, id } : null;
|
|
}
|
|
target = (work.progress as { successorId: string }).successorId;
|
|
}
|
|
throw new Error('Cyclic operations work successor');
|
|
}
|
|
|
|
async fail(attempt: OperationAttempt, invalidToken: boolean) {
|
|
return this.prisma.$transaction(async (tx) => {
|
|
if (!(await claim(tx, attempt))) return false;
|
|
await tx.tbankSyncWork.update({
|
|
where: { id: attempt.id },
|
|
data: {
|
|
state: 'failed',
|
|
errorCode: invalidToken ? TbankErrorCode.InvalidToken : TbankErrorCode.Unavailable,
|
|
},
|
|
});
|
|
if (invalidToken) {
|
|
await tx.tbankConnection.update({
|
|
where: {
|
|
userId_revision: {
|
|
userId: attempt.userId,
|
|
revision: attempt.revision,
|
|
},
|
|
},
|
|
data: { invalidToken: true },
|
|
});
|
|
}
|
|
return true;
|
|
});
|
|
}
|
|
|
|
async stagePage(
|
|
attempt: OperationAttempt,
|
|
interval: OperationInterval,
|
|
cursor: string | undefined,
|
|
response: TbankOperationsByCursorResponse,
|
|
): Promise<boolean> {
|
|
const sourceKey = intervalKey(interval);
|
|
return this.prisma.$transaction(async (tx) => {
|
|
if (!(await claim(tx, attempt))) return false;
|
|
const stage = {
|
|
workId: attempt.id,
|
|
attemptId: attempt.attemptId,
|
|
sourceKey,
|
|
};
|
|
const marker = await tx.tbankSyncStage.findUnique({
|
|
where: {
|
|
workId_attemptId_sourceKey_itemKey: {
|
|
...stage,
|
|
itemKey: progressItem,
|
|
},
|
|
},
|
|
});
|
|
const progress = marker?.payload as PageProgress | undefined;
|
|
if (progress?.complete || (progress?.cursor ?? '') !== (cursor ?? '')) {
|
|
throw new Error('Unexpected operations page');
|
|
}
|
|
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()) {
|
|
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;
|
|
// Provider pagination position is not part of a financial row.
|
|
const identity = { ...stage, itemKey: operation.id };
|
|
const previous = await tx.tbankSyncStage.findUnique({
|
|
where: { workId_attemptId_sourceKey_itemKey: identity },
|
|
});
|
|
if (previous && !isDeepStrictEqual(previous.payload, payload)) {
|
|
throw new Error('Conflicting broker operation identity');
|
|
}
|
|
if (!previous) {
|
|
await tx.tbankSyncStage.create({ data: { ...identity, payload } });
|
|
}
|
|
}
|
|
const payload: PageProgress = {
|
|
complete: !response.hasNext,
|
|
cursor: nextCursor,
|
|
seen: [
|
|
...(progress?.seen ?? []),
|
|
...(response.hasNext ? [nextCursor] : []),
|
|
],
|
|
pages: (progress?.pages ?? INITIAL_PAGE_COUNT) + PAGE_INCREMENT,
|
|
};
|
|
await tx.tbankSyncStage.upsert({
|
|
where: {
|
|
workId_attemptId_sourceKey_itemKey: {
|
|
...stage,
|
|
itemKey: progressItem,
|
|
},
|
|
},
|
|
create: { ...stage, itemKey: progressItem, payload },
|
|
update: { payload },
|
|
});
|
|
await tx.tbankSyncWork.update({
|
|
where: { id: attempt.id },
|
|
data: { progress: { sourceKey, ...payload } },
|
|
});
|
|
return true;
|
|
});
|
|
}
|
|
|
|
// A complete shown scope passes every account/month/comparison here together.
|
|
// Initial missing history may pass one completed interval at a time.
|
|
async publish(
|
|
attempt: OperationAttempt,
|
|
intervals: OperationInterval[],
|
|
fetchedAt: Date,
|
|
advanceWork = false,
|
|
): Promise<boolean> {
|
|
if (!intervals.length) {
|
|
throw new Error('Empty operations publication group');
|
|
}
|
|
return this.prisma.$transaction(
|
|
async (tx) => {
|
|
if (!(await claim(tx, attempt))) return false;
|
|
const changedAccounts = new Set<string>();
|
|
const groupRows = new Map<string, Prisma.JsonValue>();
|
|
for (const interval of intervals) {
|
|
const sourceKey = intervalKey(interval);
|
|
const stage = {
|
|
workId: attempt.id,
|
|
attemptId: attempt.attemptId,
|
|
sourceKey,
|
|
};
|
|
const marker = await tx.tbankSyncStage.findUnique({
|
|
where: {
|
|
workId_attemptId_sourceKey_itemKey: {
|
|
...stage,
|
|
itemKey: progressItem,
|
|
},
|
|
},
|
|
});
|
|
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 } },
|
|
});
|
|
for (const row of staged) {
|
|
const identity = JSON.stringify([interval.accountId, row.itemKey]);
|
|
const previous = groupRows.get(identity);
|
|
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) => {
|
|
const state = (row.payload as TbankOperationItem).state;
|
|
return !state || isTbankProtocolValue(state, TbankOperationState.Executed);
|
|
});
|
|
const scope = {
|
|
userId: attempt.userId,
|
|
revision: attempt.revision,
|
|
accountId: interval.accountId,
|
|
};
|
|
const existing = await tx.tbankOperation.findMany({
|
|
where: {
|
|
...scope,
|
|
occurredAt: { gte: interval.from, lt: interval.to },
|
|
...(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) {
|
|
await tx.tbankOperation.delete({
|
|
where: {
|
|
userId_revision_accountId_brokerOperationId: {
|
|
...scope,
|
|
brokerOperationId: row.brokerOperationId,
|
|
},
|
|
},
|
|
});
|
|
}
|
|
if (removed.length) changedAccounts.add(interval.accountId);
|
|
for (const row of rows) {
|
|
const operation = row.payload as TbankOperationItem;
|
|
const identity = { ...scope, brokerOperationId: row.itemKey };
|
|
const previous = await tx.tbankOperation.findUnique({
|
|
where: { userId_revision_accountId_brokerOperationId: identity },
|
|
});
|
|
if (previous && isDeepStrictEqual(previous.payload, operation)) {
|
|
continue;
|
|
}
|
|
const data = {
|
|
occurredAt: operationTimestamp(operation.date),
|
|
instrumentUid: operation.instrumentUid ?? null,
|
|
category: operationCategory(operation.type),
|
|
payload: row.payload as Prisma.InputJsonObject,
|
|
};
|
|
await tx.tbankOperation.upsert({
|
|
where: { userId_revision_accountId_brokerOperationId: identity },
|
|
create: { ...identity, ...data },
|
|
update: data,
|
|
});
|
|
changedAccounts.add(interval.accountId);
|
|
}
|
|
const coverage = {
|
|
...scope,
|
|
instrumentPredicate: interval.instrumentUid ?? '',
|
|
from: interval.from,
|
|
to: interval.to,
|
|
};
|
|
await tx.tbankOperationCoverage.upsert({
|
|
where: {
|
|
userId_revision_accountId_instrumentPredicate_from_to: coverage,
|
|
},
|
|
create: { ...coverage, fetchedAt, asOf: interval.to },
|
|
update: { fetchedAt, asOf: interval.to },
|
|
});
|
|
}
|
|
for (const accountId of changedAccounts) {
|
|
await tx.tbankAccount.update({
|
|
where: {
|
|
userId_revision_accountId: {
|
|
userId: attempt.userId,
|
|
revision: attempt.revision,
|
|
accountId,
|
|
},
|
|
},
|
|
data: { operationsRevision: { increment: 1 } },
|
|
});
|
|
}
|
|
for (const interval of intervals) {
|
|
await tx.tbankSyncStage.deleteMany({
|
|
where: {
|
|
workId: attempt.id,
|
|
attemptId: attempt.attemptId,
|
|
sourceKey: intervalKey(interval),
|
|
},
|
|
});
|
|
}
|
|
if (advanceWork) {
|
|
const work = await tx.tbankSyncWork.findUniqueOrThrow({
|
|
where: { id: attempt.id },
|
|
});
|
|
const intent = work.intent as OperationsIntent;
|
|
const published = new Set(intervals.map(intervalKey));
|
|
const remaining = intent.intervals.filter(
|
|
(key) => !published.has(key),
|
|
);
|
|
await tx.tbankSyncWork.update({
|
|
where: { id: attempt.id },
|
|
data: {
|
|
intent: {
|
|
...intent,
|
|
intervals: remaining,
|
|
atomic:
|
|
intent.atomic &&
|
|
(intent.atomicIntervals ?? intent.intervals).some(
|
|
(key) => !published.has(key),
|
|
),
|
|
atomicIntervals: (
|
|
intent.atomicIntervals ?? intent.intervals
|
|
).filter((key) => !published.has(key)),
|
|
},
|
|
state: remaining.length ? 'running' : 'complete',
|
|
priority: remaining.some((key) =>
|
|
(
|
|
intent.screenIntervals ??
|
|
(work.priority === operationsSyncPriority.screen ? intent.intervals : [])
|
|
).includes(key),
|
|
)
|
|
? operationsSyncPriority.screen
|
|
: operationsSyncPriority.history,
|
|
progress: { remainingIntervals: remaining.length },
|
|
},
|
|
});
|
|
}
|
|
return true;
|
|
},
|
|
{ timeout: 30_000 },
|
|
);
|
|
}
|
|
}
|
|
|
|
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,
|
|
): Promise<boolean> {
|
|
const result = await tx.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: false },
|
|
},
|
|
data: { updatedAt: new Date() },
|
|
});
|
|
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 } : {}),
|
|
};
|
|
}
|