feat!: persist T-Bank data and synchronize reads
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Failing after 10m45s
Conventional commits / commits (pull_request) Successful in 29s
Backend security / security (pull_request) Successful in 8s
Backend verify / verify (pull_request) Failing after 10m37s
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Failing after 10m45s
Conventional commits / commits (pull_request) Successful in 29s
Backend security / security (pull_request) Successful in 8s
Backend verify / verify (pull_request) Failing after 10m37s
Store account membership, portfolio, operations, forecasts and market sources in SQLite. Coalesce scoped refreshes, fence connection revisions, preserve complete financial snapshots and stabilize operation period state. BREAKING CHANGE: investment reads return data/meta resource envelopes and locally persisted snapshots; consumers must use the matching OpenAPI contract and explicit synchronization endpoints.
This commit is contained in:
@@ -0,0 +1,922 @@
|
||||
import {
|
||||
BadRequestException,
|
||||
Injectable,
|
||||
NotFoundException,
|
||||
} from '@nestjs/common';
|
||||
import { createHash, randomUUID } from 'node:crypto';
|
||||
import { isDeepStrictEqual } from 'node:util';
|
||||
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
|
||||
import { PrismaService } from '../../../../prisma/prisma.service';
|
||||
import type { Prisma } from '../../../../../generated/prisma/client';
|
||||
import type {
|
||||
TbankOperationItem,
|
||||
TbankOperationsByCursorResponse,
|
||||
} from '../../../../integrations/tbank/tbank.api';
|
||||
import { DateOnlyRange, dateOnlyBounds } from '../../investment-calendar';
|
||||
import { operationCategory } from './brokerage-operation.mapper';
|
||||
import {
|
||||
type OperationInterval,
|
||||
operationIntervals,
|
||||
operationIntervalsOverlap,
|
||||
uncoveredOperationIntervals,
|
||||
} from '../domain/operations-sync-plan';
|
||||
export type { OperationInterval } from '../domain/operations-sync-plan';
|
||||
|
||||
export type OperationAttempt = {
|
||||
id: string;
|
||||
attemptId: string;
|
||||
userId: string;
|
||||
revision: string;
|
||||
};
|
||||
type OperationsIntent = {
|
||||
intervals: string[];
|
||||
atomic: boolean;
|
||||
atomicIntervals?: string[];
|
||||
screenIntervals?: string[];
|
||||
};
|
||||
type PageProgress = {
|
||||
complete: boolean;
|
||||
cursor: string;
|
||||
seen: string[];
|
||||
pages: number;
|
||||
};
|
||||
const progressItem = '';
|
||||
const NANOSECONDS_PER_MILLISECOND = 1_000_000;
|
||||
|
||||
@Injectable()
|
||||
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[];
|
||||
limit?: number;
|
||||
periods?: (openedAt: Date | undefined) => DateOnlyRange[];
|
||||
filterKey: string;
|
||||
cursor?: { binding: string; at: string; accountId: string; id: 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;
|
||||
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 bounds = dateOnlyBounds(period);
|
||||
// Date-only end is inclusive; persisted coverage is half-open.
|
||||
const end = new Date(`${period.to}T21:00:00Z`);
|
||||
return {
|
||||
...bounds,
|
||||
key: `${period.from}/${period.to}`,
|
||||
movingEnd: end.valueOf() > Date.now(),
|
||||
to: new Date(Math.min(end.valueOf(), Date.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: 'INVALID_OPERATIONS_CURSOR' });
|
||||
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 + 1 }),
|
||||
});
|
||||
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')
|
||||
.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 > 1) {
|
||||
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 === 0,
|
||||
);
|
||||
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 <= 1 &&
|
||||
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: {
|
||||
atomic: previousIntent?.atomic ?? publicationAtomic,
|
||||
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]
|
||||
.filter((item) =>
|
||||
intervals.some((target) =>
|
||||
operationIntervalsOverlap(item, target),
|
||||
),
|
||||
)
|
||||
.map(intervalKey)
|
||||
: []),
|
||||
]),
|
||||
],
|
||||
intervals: (priority < previous.priority
|
||||
? [
|
||||
...pending.filter((item) =>
|
||||
intervals.some((target) =>
|
||||
operationIntervalsOverlap(item, target),
|
||||
),
|
||||
),
|
||||
...additions,
|
||||
...pending.filter(
|
||||
(item) =>
|
||||
!intervals.some((target) =>
|
||||
operationIntervalsOverlap(item, target),
|
||||
),
|
||||
),
|
||||
]
|
||||
: [...pending, ...additions]
|
||||
).map(intervalKey),
|
||||
},
|
||||
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 <= 1
|
||||
? 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 === 1 &&
|
||||
intervals.some((incoming) =>
|
||||
operationIntervalsOverlap(target, incoming),
|
||||
)) ||
|
||||
(
|
||||
previousIntent?.screenIntervals ??
|
||||
(previous?.priority === 1
|
||||
? (previousIntent?.intervals ?? [])
|
||||
: [])
|
||||
).some((key) =>
|
||||
operationIntervalsOverlap(target, decodeInterval(key)),
|
||||
),
|
||||
)
|
||||
.map(intervalKey),
|
||||
atomic: atomicTargets.length > 0,
|
||||
};
|
||||
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 ? 'INVALID_TBANK_TOKEN' : 'TBANK_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 = new Date(
|
||||
Number(operation.date?.seconds) * MILLISECONDS_PER_SECOND +
|
||||
Math.floor(
|
||||
(operation.date?.nanos ?? 0) / NANOSECONDS_PER_MILLISECOND,
|
||||
),
|
||||
);
|
||||
if (!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 ?? 0) + 1,
|
||||
};
|
||||
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 || state === 'OPERATION_STATE_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: new Date(
|
||||
Number(operation.date?.seconds) * MILLISECONDS_PER_SECOND +
|
||||
Math.floor(
|
||||
(operation.date?.nanos ?? 0) / NANOSECONDS_PER_MILLISECOND,
|
||||
),
|
||||
),
|
||||
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 === 1 ? intent.intervals : [])
|
||||
).includes(key),
|
||||
)
|
||||
? 1
|
||||
: 2,
|
||||
progress: { remainingIntervals: remaining.length },
|
||||
},
|
||||
});
|
||||
}
|
||||
return true;
|
||||
},
|
||||
{ timeout: 30_000 },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function intervalKey(interval: OperationInterval): string {
|
||||
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 === 1;
|
||||
}
|
||||
|
||||
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 } : {}),
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user