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