ksv741
9cc9fbdcff
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Failing after 2m17s
Conventional commits / commits (pull_request) Successful in 31s
Backend security / security (pull_request) Successful in 8s
Backend verify / verify (pull_request) Failing after 1m56s
536 lines
17 KiB
TypeScript
536 lines
17 KiB
TypeScript
import {
|
|
Injectable,
|
|
NotFoundException,
|
|
ServiceUnavailableException,
|
|
} from '@nestjs/common';
|
|
import { randomUUID } from 'node:crypto';
|
|
|
|
import { formatDecimal } from '../../../../common/decimal';
|
|
import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
|
|
import { MILLISECONDS_PER_DAY } from '../../../../common/time';
|
|
import { decimal } from '../../../../integrations/tbank/decimal';
|
|
import {
|
|
TbankAccessService,
|
|
TbankClient,
|
|
type TbankShare,
|
|
type TbankSyncAttempt,
|
|
} from '../../../../integrations/tbank/tbank.api';
|
|
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
|
import { planInvestmentScreen } from '../../investment-sync-plan';
|
|
import {
|
|
candleGaps,
|
|
type CandleRange,
|
|
candleWindows,
|
|
utcDay,
|
|
} from '../domain/candle-coverage';
|
|
import { CandleRepository } from '../infrastructure/candle.repository';
|
|
import { CatalogEnrichmentService } from './catalog-enrichment.service';
|
|
import { timestamp } from './instrument-market';
|
|
|
|
const FIRST_BATCH_OFFSET = 0;
|
|
const BACKGROUND_SYNC_PRIORITY = 2;
|
|
const REPLACED_WINDOW_COUNT = 1;
|
|
const NEXT_WINDOW_OFFSET = 1;
|
|
|
|
type CandleInput = {
|
|
instrumentType: 'bond' | 'share';
|
|
period: 'all' | 'halfYear' | 'month' | 'year';
|
|
range?: CandleRange;
|
|
uid: string;
|
|
};
|
|
type Producer = {
|
|
attempt: TbankSyncAttempt;
|
|
force: boolean;
|
|
input: CandleInput;
|
|
now: Date;
|
|
observers: TbankSyncAttempt[];
|
|
pass?: {
|
|
force: boolean;
|
|
nextWindow: number;
|
|
now: Date;
|
|
ranges: CandleRange[];
|
|
};
|
|
priority: number;
|
|
};
|
|
const months = { month: 1, halfYear: 6, year: 12 };
|
|
const tailFreshForMs = 60_000;
|
|
const prefetchReadBatchSize = 100;
|
|
function scope(input: CandleInput) {
|
|
return `candles:${input.uid}:${input.period}${input.range ? `:${input.range.from.toISOString()}/${input.range.to.toISOString()}` : ''}`;
|
|
}
|
|
function rangeFor(
|
|
input: CandleInput,
|
|
instrument: TbankShare,
|
|
now: Date,
|
|
): CandleRange {
|
|
if (input.range) return input.range;
|
|
const from = utcDay(now);
|
|
const history = timestamp(instrument.first_1dayCandleDate);
|
|
if (input.period === 'all') {
|
|
if (!history) {
|
|
throw new ServiceUnavailableException({
|
|
code:
|
|
input.instrumentType === 'share' ?
|
|
'STOCK_HISTORY_START_UNAVAILABLE' :
|
|
'BOND_HISTORY_START_UNAVAILABLE',
|
|
});
|
|
}
|
|
from.setTime(utcDay(new Date(history)).valueOf());
|
|
} else {
|
|
from.setUTCMonth(from.getUTCMonth() - months[input.period]);
|
|
if (history && Date.parse(history) > from.valueOf()) {
|
|
from.setTime(utcDay(new Date(history)).valueOf());
|
|
}
|
|
}
|
|
return { from, to: new Date(utcDay(now).valueOf() + MILLISECONDS_PER_DAY) };
|
|
}
|
|
@Injectable()
|
|
export class CandleService {
|
|
private readonly runId = randomUUID();
|
|
private readonly active = new Set<string>();
|
|
private readonly producers = new Map<string, Producer>();
|
|
private readonly queue: Producer[] = [];
|
|
private reservations: Promise<unknown> = Promise.resolve();
|
|
private running = false;
|
|
private readonly prefetches = new Map<string, Promise<void>>();
|
|
constructor(
|
|
private readonly repository: CandleRepository,
|
|
private readonly sources: CatalogEnrichmentService,
|
|
private readonly access: TbankAccessService,
|
|
private readonly tbank: TbankClient,
|
|
private readonly logger: StructuredLogger,
|
|
) {}
|
|
|
|
async read(userId: string, input: CandleInput, observe = false) {
|
|
const { revision } = await this.access.local(userId);
|
|
const [detail] = await this.sources.readMany(
|
|
userId,
|
|
input.instrumentType,
|
|
[input.uid],
|
|
true,
|
|
['detail'],
|
|
);
|
|
if (detail.meta.availability === 'complete' && !detail.instrument) {
|
|
throw new NotFoundException();
|
|
}
|
|
const now = new Date();
|
|
const range = detail.instrument ?
|
|
rangeFor(input, detail.instrument, now) :
|
|
null;
|
|
const saved = range ? await this.repository.read(input.uid, range) : null;
|
|
const gaps = range && saved ? candleGaps(range, saved.coverage, now) : [];
|
|
if (!observe && (!saved || gaps.length)) {
|
|
await this.request(userId, revision, input, now);
|
|
}
|
|
const work = await this.repository.work(userId, revision, scope(input));
|
|
const fetchedAt =
|
|
saved?.coverage
|
|
.map((row) => row.fetchedAt)
|
|
.sort((a, b) => a.valueOf() - b.valueOf())[0] ?? null;
|
|
const covered = saved?.coverage.length ? true : false;
|
|
const completeCoverage = (saved?.coverage ?? [])
|
|
.filter((row) => row.isComplete || row.from >= utcDay(now))
|
|
.map((row) => ({ ...row, isComplete: true }));
|
|
const missing = range ? candleGaps(range, completeCoverage, now) : [];
|
|
const complete = Boolean(saved && range && !missing.length);
|
|
const publicRange = (value: CandleRange) => ({
|
|
from: value.from.toISOString(),
|
|
to: value.to.toISOString(),
|
|
});
|
|
return {
|
|
data:
|
|
!saved || !covered ?
|
|
null :
|
|
{
|
|
candles: saved.data,
|
|
currency:
|
|
input.instrumentType === 'share' ?
|
|
detail.instrument?.currency?.toUpperCase() || null :
|
|
null,
|
|
historyFrom: timestamp(detail?.instrument?.first_1dayCandleDate),
|
|
observedAt: fetchedAt?.toISOString() ?? now.toISOString(),
|
|
},
|
|
meta: investmentResourceMeta({
|
|
resourceKey: scope(input),
|
|
dataRevision: saved?.dataRevision ?? '0',
|
|
availability: !covered ? 'missing' : complete ? 'complete' : 'partial',
|
|
fetchedAt,
|
|
freshForMs: Number.POSITIVE_INFINITY,
|
|
work,
|
|
coverage: {
|
|
requested: range ? [publicRange(range)] : [],
|
|
completeRanges: range ?
|
|
completeCoverage.flatMap((row) => {
|
|
const from = new Date(
|
|
Math.max(row.from.valueOf(), range.from.valueOf()),
|
|
);
|
|
const to = new Date(
|
|
Math.min(row.to.valueOf(), range.to.valueOf()),
|
|
);
|
|
return from < to ? [publicRange({ from, to })] : [];
|
|
}) :
|
|
[],
|
|
missingRanges: missing.map(publicRange),
|
|
asOf: fetchedAt?.toISOString() ?? null,
|
|
historyBoundary: timestamp(detail?.instrument?.first_1dayCandleDate) ?
|
|
'known' :
|
|
'unknown',
|
|
historyState: complete ?
|
|
'complete' :
|
|
work?.state === 'failed' ?
|
|
'failed' :
|
|
work?.state === 'running' ?
|
|
'loading' :
|
|
covered ?
|
|
'partial' :
|
|
'missing',
|
|
},
|
|
dependencies: [
|
|
detail.meta,
|
|
...(saved?.coverage ?? []).map((row) =>
|
|
investmentResourceMeta({
|
|
resourceKey: `${scope(input)}/${row.from.toISOString()}/${row.to.toISOString()}`,
|
|
dataRevision: String(row.dataRevision),
|
|
availability: 'complete',
|
|
fetchedAt: row.fetchedAt,
|
|
freshForMs: row.isComplete ?
|
|
Number.POSITIVE_INFINITY :
|
|
tailFreshForMs,
|
|
work: null,
|
|
}),
|
|
),
|
|
],
|
|
}),
|
|
};
|
|
}
|
|
|
|
prefetch(
|
|
userId: string,
|
|
revision: string,
|
|
positions: { instrumentUid?: null | string; type: string }[],
|
|
) {
|
|
const key = JSON.stringify([
|
|
userId,
|
|
revision,
|
|
positions
|
|
.map((position) => [position.instrumentUid, position.type])
|
|
// 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.
|
|
.sort(),
|
|
]);
|
|
const existing = this.prefetches.get(key);
|
|
if (existing) return existing;
|
|
const work = this.prefetchPositions(userId, revision, positions).finally(
|
|
() => this.prefetches.delete(key),
|
|
);
|
|
this.prefetches.set(key, work);
|
|
return work;
|
|
}
|
|
|
|
private async prefetchPositions(
|
|
userId: string,
|
|
revision: string,
|
|
positions: { instrumentUid?: null | string; type: string }[],
|
|
) {
|
|
const now = new Date();
|
|
const unique = new Map(
|
|
positions
|
|
.filter(
|
|
(position) =>
|
|
position.instrumentUid && ['BOND', 'STOCK'].includes(position.type),
|
|
)
|
|
.map((position) => [position.instrumentUid ?? '', position]),
|
|
);
|
|
const details: Awaited<ReturnType<CatalogEnrichmentService['readMany']>> =
|
|
[];
|
|
const coverage: Awaited<ReturnType<CandleRepository['coverage']>> = [];
|
|
for (const instrumentType of ['share', 'bond'] as const) {
|
|
const uids = [...unique]
|
|
.filter(
|
|
([, position]) =>
|
|
position.type === (instrumentType === 'share' ? 'STOCK' : 'BOND'),
|
|
)
|
|
.map(([uid]) => uid);
|
|
for (
|
|
let offset = FIRST_BATCH_OFFSET;
|
|
offset < uids.length;
|
|
offset += prefetchReadBatchSize
|
|
) {
|
|
const batch = uids.slice(offset, offset + prefetchReadBatchSize);
|
|
details.push(
|
|
...(await this.sources.readMany(userId, instrumentType, batch, true, [
|
|
'detail',
|
|
])),
|
|
);
|
|
coverage.push(...(await this.repository.coverage(batch)));
|
|
}
|
|
}
|
|
const byUid = new Map(details.map((detail) => [detail.input.uid, detail]));
|
|
for (const [uid, position] of unique) {
|
|
const instrumentType: CandleInput['instrumentType'] =
|
|
position.type === 'STOCK' ? 'share' : 'bond';
|
|
const detail = byUid.get(uid);
|
|
const history = timestamp(detail?.instrument?.first_1dayCandleDate);
|
|
const plan = planInvestmentScreen(
|
|
{ screen: 'account' },
|
|
{
|
|
membershipKnown: true,
|
|
accounts: [],
|
|
instruments: [
|
|
{
|
|
uid,
|
|
type: position.type as 'BOND' | 'STOCK',
|
|
historyFrom: history ? new Date(history) : null,
|
|
},
|
|
],
|
|
},
|
|
now,
|
|
);
|
|
if (
|
|
!detail?.instrument ||
|
|
plan.sources.some((source) => source.kind === 'candles')
|
|
) {
|
|
const input = { uid, instrumentType, period: 'all' as const };
|
|
const range =
|
|
detail?.instrument && history ?
|
|
rangeFor(input, detail.instrument, now) :
|
|
null;
|
|
if (
|
|
!range ||
|
|
candleGaps(
|
|
range,
|
|
coverage.filter((row) => row.instrumentUid === uid),
|
|
now,
|
|
).length
|
|
) {
|
|
await this.request(userId, revision, input, now, false, BACKGROUND_SYNC_PRIORITY);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
request(
|
|
userId: string,
|
|
revision: string,
|
|
input: CandleInput,
|
|
now = new Date(),
|
|
force = false,
|
|
priority = 1,
|
|
) {
|
|
const pending = this.reservations.then(async () => {
|
|
const reserved = await this.repository.reserve(
|
|
{ userId, revision },
|
|
scope(input),
|
|
{
|
|
uid: input.uid,
|
|
instrumentType: input.instrumentType,
|
|
ranges: [],
|
|
period: input.period,
|
|
now: now.toISOString(),
|
|
force,
|
|
},
|
|
this.runId,
|
|
this.active,
|
|
priority,
|
|
);
|
|
if (!reserved) return null;
|
|
const attempt = {
|
|
userId,
|
|
revision,
|
|
id: reserved.work.id,
|
|
attemptId: reserved.work.attemptId,
|
|
};
|
|
const producer = this.producers.get(scope(input));
|
|
if (producer) {
|
|
producer.force ||= force;
|
|
producer.priority = Math.min(producer.priority, priority);
|
|
if (force && now > producer.now) producer.now = now;
|
|
}
|
|
if (reserved.launch) {
|
|
this.active.add(attempt.attemptId);
|
|
if (producer) producer.observers.push(attempt);
|
|
else {
|
|
const next = { input, attempt, now, force, priority, observers: [] };
|
|
this.producers.set(scope(input), next);
|
|
this.queue.push(next);
|
|
}
|
|
if (!this.running) {
|
|
this.running = true;
|
|
setImmediate(() => {
|
|
void this.drain();
|
|
});
|
|
}
|
|
}
|
|
return attempt.id;
|
|
});
|
|
this.reservations = pending.catch(() => undefined);
|
|
return pending;
|
|
}
|
|
|
|
private async load(producer: Producer) {
|
|
const { input, attempt } = producer;
|
|
const instrument = await this.sources.ensureDetail(
|
|
attempt.userId,
|
|
attempt.revision,
|
|
input.instrumentType,
|
|
input.uid,
|
|
);
|
|
if (!instrument) throw new NotFoundException();
|
|
let pass = producer.pass;
|
|
if (!pass || (producer.force && (!pass.force || producer.now > pass.now))) {
|
|
const { now, force } = producer;
|
|
const range = rangeFor(input, instrument, now);
|
|
const saved = await this.repository.read(input.uid, range);
|
|
pass = producer.pass = {
|
|
now,
|
|
force,
|
|
ranges: (force ? [range] : candleGaps(range, saved.coverage, now))
|
|
.flatMap(candleWindows)
|
|
.reverse(),
|
|
nextWindow: 0,
|
|
};
|
|
}
|
|
const { ranges, now, force } = pass;
|
|
const index = pass.nextWindow;
|
|
// eslint-disable-next-line security/detect-object-injection -- The index is the internal numeric window cursor.
|
|
let window = ranges[index];
|
|
if (window && !force) {
|
|
const latest = await this.repository.read(input.uid, window);
|
|
const gaps = candleGaps(window, latest.coverage, now)
|
|
.flatMap(candleWindows)
|
|
.reverse();
|
|
ranges.splice(index, REPLACED_WINDOW_COUNT, ...gaps);
|
|
// eslint-disable-next-line security/detect-object-injection -- The index is the internal numeric window cursor.
|
|
window = ranges[index];
|
|
}
|
|
if (window) {
|
|
const token = await this.access.tokenForAttempt(attempt);
|
|
if (!token) return false;
|
|
const candles = await this.tbank.getCandles(
|
|
token,
|
|
input.uid,
|
|
window.from,
|
|
new Date(Math.min(window.to.valueOf(), now.valueOf())),
|
|
);
|
|
const rows = candles.flatMap((candle) => {
|
|
const time = timestamp(candle.time);
|
|
if (!time || !candle.close) throw new Error('Invalid candle response');
|
|
return [
|
|
{
|
|
time,
|
|
close: formatDecimal(decimal(candle.close)),
|
|
isComplete: candle.isComplete === true,
|
|
},
|
|
];
|
|
});
|
|
if (!(await this.repository.stage(attempt, window, rows, now))) {
|
|
return false;
|
|
}
|
|
if (
|
|
!force &&
|
|
!(await this.repository.publish(
|
|
attempt,
|
|
{ uid: input.uid, ranges: [window] },
|
|
false,
|
|
))
|
|
) {
|
|
return false;
|
|
}
|
|
pass.nextWindow = index + NEXT_WINDOW_OFFSET;
|
|
if (pass.nextWindow < ranges.length) return null;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
private async drain() {
|
|
try {
|
|
while (this.queue.length) {
|
|
this.queue.sort((a, b) => a.priority - b.priority);
|
|
const producer = this.queue.shift();
|
|
if (!producer) break;
|
|
let success = false;
|
|
let code = 'TBANK_UNAVAILABLE';
|
|
try {
|
|
const result = await this.load(producer);
|
|
if (result === null) {
|
|
this.queue.push(producer);
|
|
continue;
|
|
}
|
|
success = result;
|
|
} catch (error) {
|
|
if (error instanceof Error && error.message === 'INVALID_TBANK_TOKEN') {
|
|
await this.access.rejectMembershipToken(producer.attempt);
|
|
}
|
|
if (error instanceof ServiceUnavailableException) {
|
|
const response = error.getResponse();
|
|
if (
|
|
typeof response === 'object' &&
|
|
'code' in response &&
|
|
typeof response.code === 'string'
|
|
) {
|
|
code = response.code;
|
|
}
|
|
}
|
|
}
|
|
// Seal publication with reservations so a late forced join cannot be
|
|
// acknowledged after the final demand check but before work completion.
|
|
const finished = this.reservations.then(async () => {
|
|
const pass = producer.pass;
|
|
if (success && pass) {
|
|
if (producer.force && (!pass.force || producer.now > pass.now)) {
|
|
return false;
|
|
}
|
|
try {
|
|
success = await this.repository.publish(producer.attempt, {
|
|
uid: producer.input.uid,
|
|
ranges: pass.force ? pass.ranges : [],
|
|
});
|
|
} catch {
|
|
success = false;
|
|
}
|
|
}
|
|
if (!success) await this.repository.fail(producer.attempt, code);
|
|
this.active.delete(producer.attempt.attemptId);
|
|
for (const observer of producer.observers) {
|
|
if (
|
|
!success ||
|
|
!(await this.repository.completeObserver(
|
|
observer,
|
|
producer.attempt,
|
|
))
|
|
) {
|
|
await this.repository.fail(observer, code);
|
|
}
|
|
this.active.delete(observer.attemptId);
|
|
}
|
|
this.producers.delete(scope(producer.input));
|
|
return true;
|
|
});
|
|
this.reservations = finished.catch(() => undefined);
|
|
if (!(await finished)) {
|
|
this.queue.push(producer);
|
|
continue;
|
|
}
|
|
if (success && !producer.force && producer.input.period !== 'all') {
|
|
await this.request(
|
|
producer.attempt.userId,
|
|
producer.attempt.revision,
|
|
{ ...producer.input, period: 'all' },
|
|
producer.now,
|
|
false,
|
|
BACKGROUND_SYNC_PRIORITY,
|
|
);
|
|
}
|
|
}
|
|
} catch {
|
|
this.logger.event('error', 'candles.persistence-failed', {
|
|
code: 'TBANK_UNAVAILABLE',
|
|
});
|
|
} finally {
|
|
this.active.clear();
|
|
this.producers.clear();
|
|
this.queue.length = 0;
|
|
this.running = false;
|
|
}
|
|
}
|
|
}
|