refactor: organize backend by business responsibility
This commit is contained in:
@@ -0,0 +1,509 @@
|
||||
import {
|
||||
BadRequestException,
|
||||
Injectable,
|
||||
ServiceUnavailableException,
|
||||
} from '@nestjs/common';
|
||||
import { LRUCache } from 'lru-cache';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { RequestContextService } from '../../../../common/logging/request-context.service';
|
||||
import {
|
||||
InstrumentMetadataService,
|
||||
TbankAccessService,
|
||||
TbankClient,
|
||||
} from '../../../../integrations/tbank/tbank.api';
|
||||
import { AccountReferenceDto } from '../../account-reference.dto';
|
||||
import { DateOnlyRange, dateOnlyBounds } from '../../investment-calendar';
|
||||
import {
|
||||
BrokerageOperationDto,
|
||||
BrokerageOperationsPageDto,
|
||||
BrokerageOperationsQueryDto,
|
||||
} from '../presentation/brokerage-operation.dto';
|
||||
import {
|
||||
mapBrokerageOperation,
|
||||
operationCategory,
|
||||
} from '../infrastructure/brokerage-operation.mapper';
|
||||
|
||||
type OperationStream = {
|
||||
account: AccountReferenceDto;
|
||||
pending: BrokerageOperationDto[];
|
||||
cursor?: string;
|
||||
finished: boolean;
|
||||
seen: string[];
|
||||
lastOccurredAt?: number;
|
||||
};
|
||||
type AllOperationsContinuation = {
|
||||
signature: string;
|
||||
range: { from: Date; to: Date };
|
||||
streams: OperationStream[];
|
||||
page?: Promise<BrokerageOperationsPageDto>;
|
||||
};
|
||||
|
||||
const freshForMs = 30_000;
|
||||
const maxCachedPages = 100;
|
||||
|
||||
@Injectable()
|
||||
export class BrokerageOperationsService {
|
||||
private readonly pages = new LRUCache<string, BrokerageOperationsPageDto>({
|
||||
max: maxCachedPages,
|
||||
ttl: freshForMs,
|
||||
ttlAutopurge: true,
|
||||
});
|
||||
|
||||
private readonly continuations = new LRUCache<
|
||||
string,
|
||||
AllOperationsContinuation
|
||||
>({
|
||||
max: 100,
|
||||
ttl: 5 * 60_000,
|
||||
ttlAutopurge: true,
|
||||
});
|
||||
|
||||
constructor(
|
||||
private readonly access: TbankAccessService,
|
||||
private readonly tbank: TbankClient,
|
||||
private readonly requestContext: RequestContextService,
|
||||
private readonly metadata: InstrumentMetadataService,
|
||||
) {}
|
||||
|
||||
async get(
|
||||
userId: string,
|
||||
accountId: string,
|
||||
query: BrokerageOperationsQueryDto,
|
||||
): Promise<BrokerageOperationsPageDto> {
|
||||
this.requestContext.set({ userId, brokerageAccountId: accountId });
|
||||
const range = this.range(query);
|
||||
|
||||
return this.access.withAccount(
|
||||
userId,
|
||||
accountId,
|
||||
({ token, generation, account }) =>
|
||||
this.page(
|
||||
token,
|
||||
generation,
|
||||
userId,
|
||||
accountId,
|
||||
query,
|
||||
query.instrumentUid && !query.from
|
||||
? {
|
||||
...range,
|
||||
from: new Date(Number(account.openedDate?.seconds ?? 0) * 1000),
|
||||
}
|
||||
: range,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
async getAll(
|
||||
userId: string,
|
||||
query: BrokerageOperationsQueryDto,
|
||||
): Promise<BrokerageOperationsPageDto> {
|
||||
return this.access.withSupportedAccounts(
|
||||
userId,
|
||||
async ({ token, generation, accounts }) => {
|
||||
const signature = JSON.stringify({
|
||||
userId,
|
||||
generation,
|
||||
accounts: accounts.map((account) => account.id).sort(),
|
||||
instrumentUid: query.instrumentUid ?? null,
|
||||
from: query.from ?? null,
|
||||
to: query.to ?? null,
|
||||
categories: [...new Set(query.categories ?? [])].sort(),
|
||||
limit: query.limit ?? 100,
|
||||
});
|
||||
const continuation = query.cursor
|
||||
? this.continuations.get(query.cursor)
|
||||
: undefined;
|
||||
if (
|
||||
query.cursor &&
|
||||
(!continuation || continuation.signature !== signature)
|
||||
)
|
||||
throw new BadRequestException({ code: 'INVALID_OPERATIONS_CURSOR' });
|
||||
if (continuation?.page) return continuation.page;
|
||||
const oldest = accounts
|
||||
.map((account) => account.openedOn)
|
||||
.filter((date): date is string => Boolean(date))
|
||||
.sort()[0];
|
||||
const range =
|
||||
continuation?.range ??
|
||||
this.range({
|
||||
...query,
|
||||
from:
|
||||
query.from ??
|
||||
(query.instrumentUid
|
||||
? ((accounts.every((account) => account.openedOn)
|
||||
? oldest
|
||||
: undefined) ?? '1970-01-01')
|
||||
: undefined),
|
||||
});
|
||||
const streams: OperationStream[] = continuation
|
||||
? continuation.streams.map((stream) => ({
|
||||
...stream,
|
||||
pending: [...stream.pending],
|
||||
seen: [...stream.seen],
|
||||
}))
|
||||
: accounts.map(({ id, name }) => ({
|
||||
account: { id, name },
|
||||
pending: [],
|
||||
finished: false,
|
||||
seen: [],
|
||||
}));
|
||||
const result = this.mergePage(
|
||||
token,
|
||||
generation,
|
||||
userId,
|
||||
query,
|
||||
signature,
|
||||
range,
|
||||
streams,
|
||||
);
|
||||
if (continuation) continuation.page = result;
|
||||
try {
|
||||
return await result;
|
||||
} catch (error) {
|
||||
if (continuation) continuation.page = undefined;
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private async mergePage(
|
||||
token: string,
|
||||
generation: string,
|
||||
userId: string,
|
||||
query: BrokerageOperationsQueryDto,
|
||||
signature: string,
|
||||
range: { from: Date; to: Date },
|
||||
streams: OperationStream[],
|
||||
): Promise<BrokerageOperationsPageDto> {
|
||||
const items: BrokerageOperationDto[] = [];
|
||||
// Each stream retains unconsumed rows. A page boundary never discards them.
|
||||
while (items.length < (query.limit ?? 100)) {
|
||||
for (const stream of streams) {
|
||||
while (!stream.pending.length && !stream.finished) {
|
||||
const page = await this.page(
|
||||
token,
|
||||
generation,
|
||||
userId,
|
||||
stream.account.id,
|
||||
{ ...query, cursor: stream.cursor },
|
||||
range,
|
||||
);
|
||||
for (const item of page.items) {
|
||||
const time = Date.parse(item.occurredAt);
|
||||
if (
|
||||
!Number.isFinite(time) ||
|
||||
(stream.lastOccurredAt !== undefined &&
|
||||
time > stream.lastOccurredAt)
|
||||
)
|
||||
throw new ServiceUnavailableException({
|
||||
code: 'TBANK_UNAVAILABLE',
|
||||
});
|
||||
stream.lastOccurredAt = time;
|
||||
}
|
||||
if (page.hasNext) {
|
||||
if (!page.nextCursor || stream.seen.includes(page.nextCursor))
|
||||
throw new ServiceUnavailableException({
|
||||
code: 'TBANK_UNAVAILABLE',
|
||||
});
|
||||
stream.seen.push(page.nextCursor);
|
||||
stream.cursor = page.nextCursor;
|
||||
} else stream.finished = true;
|
||||
stream.pending = page.items.map((item) => ({
|
||||
...item,
|
||||
account: stream.account,
|
||||
}));
|
||||
}
|
||||
}
|
||||
const next = streams
|
||||
.filter((stream) => stream.pending.length)
|
||||
.sort(
|
||||
(left, right) =>
|
||||
Date.parse(right.pending[0].occurredAt) -
|
||||
Date.parse(left.pending[0].occurredAt) ||
|
||||
left.account.id.localeCompare(right.account.id),
|
||||
)[0];
|
||||
if (!next) break;
|
||||
items.push(...next.pending.splice(0, 1));
|
||||
}
|
||||
const hasNext = streams.some(
|
||||
(stream) => stream.pending.length || !stream.finished,
|
||||
);
|
||||
const nextCursor = hasNext ? randomUUID() : null;
|
||||
if (nextCursor)
|
||||
this.continuations.set(nextCursor, { signature, range, streams });
|
||||
return {
|
||||
accountId: 'all',
|
||||
items,
|
||||
hasNext,
|
||||
nextCursor,
|
||||
observedAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
private async collectAllRanges(
|
||||
userId: string,
|
||||
ranges:
|
||||
| DateOnlyRange[]
|
||||
| ((accountOpenedAt: Date | undefined) => DateOnlyRange[]),
|
||||
): Promise<{
|
||||
accountOpenedAt: Date | undefined;
|
||||
pages: BrokerageOperationsPageDto[][];
|
||||
}> {
|
||||
return this.access.withSupportedAccounts(
|
||||
userId,
|
||||
async ({ token, generation, accounts }) => {
|
||||
// The current cohort includes newly opened accounts with no earlier receipts.
|
||||
const oldest = accounts
|
||||
.map((account) => account.openedOn)
|
||||
.filter((date): date is string => Boolean(date))
|
||||
.sort()[0];
|
||||
const accountOpenedAt =
|
||||
oldest && accounts.every((account) => account.openedOn)
|
||||
? new Date(`${oldest}T00:00:00+03:00`)
|
||||
: undefined;
|
||||
const selected =
|
||||
typeof ranges === 'function' ? ranges(accountOpenedAt) : ranges;
|
||||
const pages: {
|
||||
items: (BrokerageOperationDto & { account: AccountReferenceDto })[];
|
||||
}[][] = selected.map(() => []);
|
||||
for (const account of accounts) {
|
||||
for (const [index, range] of selected.entries()) {
|
||||
const collected = await this.collectRange(
|
||||
token,
|
||||
generation,
|
||||
userId,
|
||||
account.id,
|
||||
range,
|
||||
);
|
||||
pages[index].push(
|
||||
...collected.map((page) => ({
|
||||
...page,
|
||||
items: page.items.map((item) => ({
|
||||
...item,
|
||||
account: { id: account.id, name: account.name },
|
||||
})),
|
||||
})),
|
||||
);
|
||||
}
|
||||
}
|
||||
return {
|
||||
accountOpenedAt,
|
||||
pages: pages.map((group) => [
|
||||
{
|
||||
...emptyPage('all'),
|
||||
items: group
|
||||
.flatMap((page) => page.items)
|
||||
.sort(
|
||||
(left, right) =>
|
||||
Date.parse(right.occurredAt) -
|
||||
Date.parse(left.occurredAt) ||
|
||||
left.account.id.localeCompare(right.account.id),
|
||||
),
|
||||
},
|
||||
]),
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
async collectRanges(
|
||||
userId: string,
|
||||
accountId: string,
|
||||
ranges:
|
||||
| DateOnlyRange[]
|
||||
| ((accountOpenedAt: Date | undefined) => DateOnlyRange[]),
|
||||
): Promise<{
|
||||
accountOpenedAt: Date | undefined;
|
||||
pages: BrokerageOperationsPageDto[][];
|
||||
}> {
|
||||
if (accountId === 'all') return this.collectAllRanges(userId, ranges);
|
||||
this.requestContext.set({ userId, brokerageAccountId: accountId });
|
||||
return this.access.withAccount(
|
||||
userId,
|
||||
accountId,
|
||||
async ({ token, generation, account }) => {
|
||||
const accountOpenedAt = account.openedDate?.seconds
|
||||
? new Date(Number(account.openedDate.seconds) * 1000)
|
||||
: undefined;
|
||||
const selected =
|
||||
typeof ranges === 'function' ? ranges(accountOpenedAt) : ranges;
|
||||
return {
|
||||
accountOpenedAt,
|
||||
pages: await Promise.all(
|
||||
selected.map((range) =>
|
||||
this.collectRange(token, generation, userId, accountId, range),
|
||||
),
|
||||
),
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private async collectRange(
|
||||
token: string,
|
||||
generation: string,
|
||||
userId: string,
|
||||
accountId: string,
|
||||
range: DateOnlyRange,
|
||||
): Promise<BrokerageOperationsPageDto[]> {
|
||||
const pages: BrokerageOperationsPageDto[] = [];
|
||||
const seen = new Set<string>();
|
||||
let cursor: string | undefined;
|
||||
do {
|
||||
const page = await this.page(
|
||||
token,
|
||||
generation,
|
||||
userId,
|
||||
accountId,
|
||||
{
|
||||
from: range.from,
|
||||
cursor,
|
||||
limit: 1000,
|
||||
categories: ['income'],
|
||||
},
|
||||
dateOnlyBounds(range),
|
||||
);
|
||||
pages.push(page);
|
||||
if (!page.hasNext) return pages;
|
||||
if (!page.nextCursor || seen.has(page.nextCursor))
|
||||
throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
|
||||
seen.add(page.nextCursor);
|
||||
cursor = page.nextCursor;
|
||||
} while (cursor);
|
||||
return pages;
|
||||
}
|
||||
|
||||
private async page(
|
||||
token: string,
|
||||
generation: string,
|
||||
userId: string,
|
||||
accountId: string,
|
||||
query: BrokerageOperationsQueryDto,
|
||||
range: { from: Date; to: Date },
|
||||
): Promise<BrokerageOperationsPageDto> {
|
||||
const key = cacheKey(generation, userId, accountId, query, range);
|
||||
const cached = this.pages.get(key);
|
||||
if (cached) return cached;
|
||||
if (range.from > range.to) {
|
||||
const page = emptyPage(accountId);
|
||||
this.pages.set(key, page);
|
||||
return page;
|
||||
}
|
||||
const response = await this.tbank.getOperationsByCursor(token, {
|
||||
accountId,
|
||||
from: seconds(range.from),
|
||||
to: seconds(range.to, /^\d{4}-\d{2}-\d{2}$/.test(query.to ?? '')),
|
||||
cursor: query.cursor,
|
||||
limit: query.limit ?? 100,
|
||||
state: 'OPERATION_STATE_EXECUTED',
|
||||
...(query.instrumentUid ? { instrumentId: query.instrumentUid } : {}),
|
||||
});
|
||||
const items = (response.items ?? []).filter(
|
||||
(item) =>
|
||||
(!query.instrumentUid || item.instrumentUid === query.instrumentUid) &&
|
||||
(!query.categories?.length ||
|
||||
query.categories.includes(operationCategory(item.type))),
|
||||
);
|
||||
const logoNames = new Map<string, Promise<string | null>>();
|
||||
const page = {
|
||||
accountId,
|
||||
items: await Promise.all(
|
||||
items.map(async (item) => ({
|
||||
...mapBrokerageOperation(item),
|
||||
logoName: await this.logoName(
|
||||
token,
|
||||
item.instrumentUid,
|
||||
item.figi,
|
||||
logoNames,
|
||||
),
|
||||
})),
|
||||
),
|
||||
nextCursor: response.nextCursor || null,
|
||||
hasNext: Boolean(response.hasNext),
|
||||
observedAt: new Date().toISOString(),
|
||||
};
|
||||
this.pages.set(key, page);
|
||||
return page;
|
||||
}
|
||||
|
||||
private range(query: BrokerageOperationsQueryDto) {
|
||||
const now = new Date();
|
||||
return {
|
||||
from:
|
||||
dateBoundary(query.from, false) ??
|
||||
new Date(Date.UTC(now.getUTCFullYear(), 0, 1)),
|
||||
to: dateBoundary(query.to, true) ?? now,
|
||||
};
|
||||
}
|
||||
|
||||
private async logoName(
|
||||
token: string,
|
||||
instrumentUid: string | undefined,
|
||||
figi: string | undefined,
|
||||
logoNames: Map<string, Promise<string | null>>,
|
||||
): Promise<string | null> {
|
||||
const uid = instrumentUid?.trim() || null;
|
||||
const instrumentFigi = figi?.trim() || null;
|
||||
const key =
|
||||
uid || instrumentFigi
|
||||
? JSON.stringify({ uid, figi: instrumentFigi })
|
||||
: null;
|
||||
if (!key) return Promise.resolve(null);
|
||||
const existing = logoNames.get(key);
|
||||
if (existing) return existing;
|
||||
const logoName = this.metadata
|
||||
.resolve(token, uid, instrumentFigi)
|
||||
.then((instrument) => instrument?.brand?.logoName?.trim() || null);
|
||||
logoNames.set(key, logoName);
|
||||
return logoName;
|
||||
}
|
||||
}
|
||||
|
||||
function cacheKey(
|
||||
generation: string,
|
||||
userId: string,
|
||||
accountId: string,
|
||||
query: BrokerageOperationsQueryDto,
|
||||
range: { from: Date; to: Date },
|
||||
): string {
|
||||
const categories = [...new Set(query.categories ?? [])].sort();
|
||||
const end = query.to
|
||||
? range.to.toISOString()
|
||||
: Math.floor(range.to.valueOf() / freshForMs);
|
||||
return JSON.stringify({
|
||||
generation,
|
||||
userId,
|
||||
accountId,
|
||||
from: range.from.toISOString(),
|
||||
to: end,
|
||||
categories,
|
||||
instrumentUid: query.instrumentUid ?? null,
|
||||
cursor: query.cursor ?? null,
|
||||
limit: query.limit ?? 100,
|
||||
});
|
||||
}
|
||||
|
||||
function dateBoundary(value: string | undefined, end: boolean): Date | null {
|
||||
if (!value) return null;
|
||||
if (/^\d{4}-\d{2}-\d{2}$/.test(value))
|
||||
return new Date(`${value}T${end ? '23:59:59.999' : '00:00:00.000'}Z`);
|
||||
return new Date(value);
|
||||
}
|
||||
|
||||
function seconds(value: Date, endOfSecond = false) {
|
||||
return {
|
||||
seconds: Math.floor(value.valueOf() / 1000).toString(),
|
||||
...(endOfSecond ? { nanos: 999_999_999 } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function emptyPage(accountId: string): BrokerageOperationsPageDto {
|
||||
return {
|
||||
accountId,
|
||||
items: [],
|
||||
nextCursor: null,
|
||||
hasNext: false,
|
||||
observedAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user