410 lines
13 KiB
TypeScript
410 lines
13 KiB
TypeScript
import {
|
|
Injectable,
|
|
NotFoundException,
|
|
ServiceUnavailableException,
|
|
} from '@nestjs/common';
|
|
import { randomUUID } from 'node:crypto';
|
|
|
|
import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
|
|
import { MILLISECONDS_PER_DAY } from '../../../../common/time';
|
|
import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
|
|
import {
|
|
TbankAccessService,
|
|
TbankClient,
|
|
type TbankShare,
|
|
type TbankSyncAttempt,
|
|
} from '../../../../integrations/tbank/tbank.api';
|
|
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
|
import {
|
|
candleGaps,
|
|
type CandleRange,
|
|
candleWindows,
|
|
utcDay,
|
|
} from '../domain/candle-coverage';
|
|
import { InstrumentErrorCode } from '../domain/instrument-error-code';
|
|
import { CandleRepository } from '../infrastructure/candle.repository';
|
|
import { toSavedCandle } from '../infrastructure/tbank-candle.mapper';
|
|
import { CatalogEnrichmentService } from './catalog-enrichment.service';
|
|
import { timestamp } from './instrument-market';
|
|
|
|
const REPLACED_WINDOW_COUNT = 1;
|
|
const NEXT_WINDOW_OFFSET = 1;
|
|
const NO_CANDLE_GAPS = 0;
|
|
|
|
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;
|
|
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'
|
|
? InstrumentErrorCode.StockHistoryStartUnavailable
|
|
: InstrumentErrorCode.BondHistoryStartUnavailable,
|
|
});
|
|
}
|
|
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;
|
|
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.completeness === '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 hasCompleteRange = Boolean(
|
|
range && saved && candleGaps(range, saved.coverage, now).length === NO_CANDLE_GAPS,
|
|
);
|
|
if (!observe && (!saved || !hasCompleteRange)) {
|
|
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,
|
|
}),
|
|
),
|
|
],
|
|
}),
|
|
};
|
|
}
|
|
|
|
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.map(toSavedCandle);
|
|
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: string = TbankErrorCode.Unavailable;
|
|
try {
|
|
const result = await this.load(producer);
|
|
if (result === null) {
|
|
this.queue.push(producer);
|
|
continue;
|
|
}
|
|
success = result;
|
|
} catch (error) {
|
|
if (error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)) {
|
|
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);
|
|
}
|
|
}
|
|
} catch {
|
|
this.logger.event('error', 'candles.persistence-failed', {
|
|
code: TbankErrorCode.Unavailable,
|
|
});
|
|
} finally {
|
|
this.active.clear();
|
|
this.producers.clear();
|
|
this.queue.length = 0;
|
|
this.running = false;
|
|
}
|
|
}
|
|
}
|