318 lines
12 KiB
TypeScript
318 lines
12 KiB
TypeScript
import {
|
|
Injectable,
|
|
NotFoundException,
|
|
} from '@nestjs/common';
|
|
import { randomUUID } from 'node:crypto';
|
|
|
|
import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
|
|
import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
|
|
import {
|
|
InstrumentMetadataService,
|
|
metadataKey,
|
|
TbankAccessService,
|
|
TbankClient,
|
|
TbankMembershipService,
|
|
type TbankMoney,
|
|
type TbankPortfolio,
|
|
tbankSourceTtlMs,
|
|
} from '../../../../integrations/tbank/tbank.api';
|
|
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
|
import {
|
|
hourlyAttempt,
|
|
TbankSyncRepository,
|
|
} from '../../tbank-sync/tbank-sync.api';
|
|
import { portfolioStructure } from '../domain/portfolio-structure';
|
|
import { sumMoneyByCurrency } from '../domain/portfolio-totals';
|
|
import {
|
|
portfolioFreshForMs,
|
|
PortfolioRepository,
|
|
portfolioWorkKey,
|
|
} from '../infrastructure/portfolio.repository';
|
|
import { mapPortfolio } from '../infrastructure/tbank-portfolio.mapper';
|
|
import { PortfolioDto } from '../presentation/portfolio.dto';
|
|
|
|
const MISSING_DATA_REVISION = 0;
|
|
const NO_ACCOUNTS = 0;
|
|
|
|
const membershipFreshForMs = tbankSourceTtlMs.membership;
|
|
|
|
@Injectable()
|
|
export class PortfolioService {
|
|
private readonly processRunId = randomUUID();
|
|
private readonly pending = new Map<string, Promise<void>>();
|
|
private reservations: Promise<unknown> = Promise.resolve();
|
|
|
|
constructor(
|
|
private readonly access: TbankAccessService,
|
|
private readonly tbank: TbankClient,
|
|
private readonly metadata: InstrumentMetadataService,
|
|
private readonly repository: PortfolioRepository,
|
|
private readonly membership: TbankMembershipService,
|
|
private readonly logger: StructuredLogger,
|
|
private readonly sync: TbankSyncRepository,
|
|
) {}
|
|
|
|
// Local reads never start upstream work; the hourly walk owns refreshes.
|
|
async saved(userId: string, accountId: string) {
|
|
const connection = await this.access.local(userId, accountId);
|
|
const saved = await this.repository.read(userId, connection.revision);
|
|
const hourly = await this.sync.hourly(userId);
|
|
if (
|
|
saved?.membershipAt &&
|
|
accountId !== 'all' &&
|
|
!saved.accounts.some((account) => account.accountId === accountId)
|
|
) {
|
|
throw new NotFoundException();
|
|
}
|
|
const membershipMeta = investmentResourceMeta({
|
|
resourceKey: 'accounts/membership',
|
|
dataRevision: String(saved?.membershipRevision ?? MISSING_DATA_REVISION),
|
|
availability: saved?.membershipAt ? 'complete' : 'missing',
|
|
fetchedAt: saved?.membershipAt ?? null,
|
|
freshForMs: membershipFreshForMs,
|
|
work: saved?.work.find((work) => work.scopeKey === 'membership') ?? null,
|
|
lastAttempt: hourlyAttempt(hourly, 'membership'),
|
|
});
|
|
return {
|
|
hourly,
|
|
membership: membershipMeta,
|
|
accounts: (saved?.accounts ?? [])
|
|
.filter(
|
|
(account) => accountId === 'all' || account.accountId === accountId,
|
|
)
|
|
.map((account) => ({
|
|
id: account.accountId,
|
|
name: account.name,
|
|
type: account.type,
|
|
openedOn: account.openedOn,
|
|
snapshot: account.portfolio
|
|
? {
|
|
...(account.portfolio.payload as unknown as PortfolioDto),
|
|
lastSuccessfulAt: account.portfolio.fetchedAt.toISOString(),
|
|
isStale:
|
|
Date.now() - account.portfolio.fetchedAt.valueOf() >=
|
|
portfolioFreshForMs,
|
|
}
|
|
: null,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: `portfolio/${encodeURIComponent(account.accountId)}`,
|
|
dataRevision: String(account.portfolio?.dataRevision ?? MISSING_DATA_REVISION),
|
|
availability: account.portfolio ? 'complete' : 'missing',
|
|
fetchedAt: account.portfolio?.fetchedAt ?? null,
|
|
freshForMs: portfolioFreshForMs,
|
|
work:
|
|
saved?.work.find(
|
|
(work) => work.scopeKey === portfolioWorkKey(account.accountId),
|
|
) ?? null,
|
|
lastAttempt: hourlyAttempt(hourly, 'portfolio', account.accountId),
|
|
}),
|
|
})),
|
|
};
|
|
}
|
|
|
|
async read(userId: string, accountId: string) {
|
|
const { hourly, ...saved } = await this.saved(userId, accountId);
|
|
const missing =
|
|
saved.membership.completeness === 'incomplete' ||
|
|
(saved.accounts.length > NO_ACCOUNTS && saved.accounts.some((account) => !account.snapshot));
|
|
const available =
|
|
!missing &&
|
|
(saved.membership.availability === 'available' ||
|
|
saved.accounts.some((account) => account.snapshot));
|
|
const metadata = await this.metadata.savedMany(
|
|
saved.accounts.flatMap((account) => account.snapshot?.positions ?? []),
|
|
);
|
|
const metadataDependencies = [...metadata.entries()].flatMap(
|
|
([key, source]) =>
|
|
source.complete && source.data
|
|
? [
|
|
investmentResourceMeta({
|
|
resourceKey: `instrument/metadata/${key}`,
|
|
dataRevision: source.dataRevision,
|
|
availability: 'complete',
|
|
fetchedAt: source.fetchedAt,
|
|
freshForMs: portfolioFreshForMs,
|
|
work: null,
|
|
}),
|
|
]
|
|
: [],
|
|
);
|
|
const portfolios = saved.accounts.flatMap((account) =>
|
|
account.snapshot
|
|
? [
|
|
{
|
|
...account.snapshot,
|
|
positions: account.snapshot.positions.map((position) => {
|
|
const shared = metadata.get(metadataKey(position))?.data;
|
|
return {
|
|
...position,
|
|
...(shared
|
|
? {
|
|
displayName:
|
|
shared.name?.trim() ||
|
|
shared.ticker?.trim() ||
|
|
position.displayName,
|
|
ticker: shared.ticker?.trim() || position.ticker,
|
|
isin: shared.isin?.trim() || position.isin,
|
|
logoName:
|
|
shared.brand?.logoName?.trim() || position.logoName,
|
|
}
|
|
: {}),
|
|
...(accountId === 'all'
|
|
? { account: { id: account.id, name: account.name } }
|
|
: {}),
|
|
};
|
|
}),
|
|
},
|
|
]
|
|
: [],
|
|
);
|
|
let availability: 'available' | 'complete' | 'missing' | 'partial';
|
|
if (available) availability = 'available';
|
|
else if (!missing) availability = 'complete';
|
|
else if (portfolios.length > NO_ACCOUNTS) availability = 'partial';
|
|
else availability = 'missing';
|
|
const lastSuccessfulAt =
|
|
// 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.
|
|
portfolios.map((portfolio) => portfolio.lastSuccessfulAt).sort()[0] ??
|
|
saved.membership.lastSuccessfulAt;
|
|
const data =
|
|
missing || !lastSuccessfulAt
|
|
? null
|
|
: {
|
|
lastSuccessfulAt,
|
|
isStale: portfolios.length
|
|
? portfolios.some((portfolio) => portfolio.isStale)
|
|
: saved.membership.freshness !== 'fresh',
|
|
totalsByCurrency: sumMoneyByCurrency(
|
|
portfolios.flatMap((portfolio) => portfolio.totalsByCurrency),
|
|
),
|
|
cashBalances: sumMoneyByCurrency(
|
|
portfolios.flatMap((portfolio) => portfolio.cashBalances),
|
|
),
|
|
positions: portfolios.flatMap((portfolio) => portfolio.positions),
|
|
structure: portfolioStructure(
|
|
portfolios.flatMap((portfolio) => portfolio.positions),
|
|
portfolios.flatMap((portfolio) => portfolio.cashBalances),
|
|
),
|
|
};
|
|
const work =
|
|
saved.accounts
|
|
.map((account) => account.meta.sync)
|
|
.find((work) => work.state === 'running') ??
|
|
saved.accounts
|
|
.map((account) => account.meta.sync)
|
|
.find((work) => work.state === 'failed') ??
|
|
saved.membership.sync;
|
|
return {
|
|
data,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: `portfolio/${encodeURIComponent(accountId)}`,
|
|
dataRevision: JSON.stringify([
|
|
saved.accounts.map((account) => [
|
|
account.id,
|
|
account.meta.dataRevision,
|
|
]),
|
|
metadataDependencies.map((dependency) => dependency.dataRevision),
|
|
]),
|
|
availability,
|
|
fetchedAt: lastSuccessfulAt ? new Date(lastSuccessfulAt) : null,
|
|
freshForMs: portfolios.length
|
|
? portfolioFreshForMs
|
|
: membershipFreshForMs,
|
|
work: work.id ? { ...work, id: work.id } : null,
|
|
lastAttempt: hourlyAttempt(hourly, 'portfolio', accountId === 'all' ? undefined : accountId),
|
|
dependencies: [
|
|
saved.membership,
|
|
...saved.accounts.map((account) => account.meta),
|
|
...metadataDependencies,
|
|
],
|
|
}),
|
|
};
|
|
}
|
|
|
|
request(userId: string, revision: string, accountId: string, force = false) {
|
|
return this.queueSnapshot(userId, revision, accountId, force);
|
|
}
|
|
|
|
private queueSnapshot(
|
|
userId: string,
|
|
revision: string,
|
|
accountId: string,
|
|
force = false,
|
|
) {
|
|
const reserved = this.reservations.then(async () => {
|
|
const requested = await this.repository.reserve(
|
|
userId,
|
|
revision,
|
|
accountId,
|
|
this.processRunId,
|
|
this.pending,
|
|
force,
|
|
);
|
|
if (!requested) return null;
|
|
if (!requested.launch) {
|
|
return {
|
|
id: requested.work.id,
|
|
done: this.pending.get(requested.work.attemptId) ?? Promise.resolve(),
|
|
};
|
|
}
|
|
const { id, attemptId } = requested.work;
|
|
const done = new Promise<void>((resolve) => setImmediate(resolve))
|
|
.then(async () => {
|
|
const attempt = { id, attemptId, userId, revision };
|
|
try {
|
|
const token = await this.access.tokenForAttempt(attempt);
|
|
if (!token) return;
|
|
const [portfolio, positions] = await Promise.all([
|
|
this.tbank.getPortfolio(token, accountId),
|
|
this.tbank.getPositions(token, accountId),
|
|
]);
|
|
const snapshot = await this.normalize(
|
|
portfolio.positions ?? [],
|
|
positions.money ?? [],
|
|
);
|
|
const published = await this.repository.publish(
|
|
attempt,
|
|
accountId,
|
|
snapshot,
|
|
);
|
|
if (published) {
|
|
for (const position of snapshot.positions) {
|
|
await this.metadata.request(userId, revision, position);
|
|
}
|
|
}
|
|
} catch (error) {
|
|
if (
|
|
error instanceof Error &&
|
|
isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)
|
|
) {
|
|
await this.access.rejectMembershipToken(attempt);
|
|
} else await this.repository.fail(attempt);
|
|
}
|
|
})
|
|
.catch(() =>
|
|
this.logger.event('error', 'tbank.portfolio.persistence_failed', {
|
|
userId,
|
|
}),
|
|
)
|
|
.finally(() => {
|
|
this.pending.delete(attemptId);
|
|
});
|
|
this.pending.set(attemptId, done);
|
|
return { id, done };
|
|
});
|
|
this.reservations = reserved.catch(() => undefined);
|
|
return reserved;
|
|
}
|
|
|
|
private async normalize(
|
|
upstream: NonNullable<TbankPortfolio['positions']>,
|
|
money: TbankMoney[],
|
|
): Promise<PortfolioDto> {
|
|
const lastSuccessfulAt = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
|
|
const savedMetadata = await this.metadata.savedMany(upstream);
|
|
return mapPortfolio(upstream, money, savedMetadata, lastSuccessfulAt);
|
|
}
|
|
}
|