482 lines
16 KiB
TypeScript
482 lines
16 KiB
TypeScript
import { Injectable, NotFoundException } from '@nestjs/common';
|
|
import { randomUUID } from 'node:crypto';
|
|
|
|
import { StructuredLogger } from '#src/common/logging/structured-logger.service';
|
|
import { isTbankErrorCode, TbankErrorCode } from '#src/integrations/tbank/tbank-error-code';
|
|
import {
|
|
TbankAccessService,
|
|
TbankClient,
|
|
TbankLastPrice,
|
|
tbankSourceNeedsRefresh,
|
|
tbankSourceTtlMs,
|
|
type TbankSyncAttempt,
|
|
} from '#src/integrations/tbank/tbank.api';
|
|
|
|
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
|
import {
|
|
CatalogDiscoveryRepository,
|
|
catalogScope,
|
|
type CatalogSource,
|
|
} from '../infrastructure/catalog-discovery.repository';
|
|
import { type CatalogPayload, CatalogSourceLoader } from './catalog-source.loader';
|
|
export type { CatalogPayload } from './catalog-source.loader';
|
|
import { catalogSearchLimit } from './catalog-selection';
|
|
import { publicInstrument } from './public-instrument';
|
|
|
|
const MISSING_DATA_REVISION = 0;
|
|
const FIRST_QUEUE_INDEX = 0;
|
|
const DEQUEUED_ITEM_COUNT = 1;
|
|
|
|
type Producer = {
|
|
attempt: TbankSyncAttempt;
|
|
input: CatalogSource;
|
|
observers: { attempt: TbankSyncAttempt; input: CatalogSource }[];
|
|
};
|
|
@Injectable()
|
|
export class CatalogEnrichmentService {
|
|
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 completions = new Map<
|
|
string,
|
|
{ attempt: TbankSyncAttempt; done: Promise<void>; resolve: () => void; workId: string }
|
|
>();
|
|
|
|
async ensureDetail(
|
|
userId: string,
|
|
revision: string,
|
|
instrumentType: 'bond' | 'share',
|
|
uid: string,
|
|
) {
|
|
const ids = await this.requestMany(
|
|
userId,
|
|
revision,
|
|
instrumentType,
|
|
[uid],
|
|
false,
|
|
['detail'],
|
|
);
|
|
await Promise.all(
|
|
[...this.completions.values()]
|
|
.filter((entry) => ids.includes(entry.workId))
|
|
.map((entry) => entry.done),
|
|
);
|
|
const rows = await this.repository.readSources([
|
|
{ instrumentType, uid, kind: 'detail' },
|
|
]);
|
|
if (!rows.length) throw new Error('Instrument dependency missing');
|
|
return (rows[0].payload as CatalogPayload).instrument ?? null;
|
|
}
|
|
|
|
constructor(
|
|
private readonly repository: CatalogDiscoveryRepository,
|
|
private readonly access: TbankAccessService,
|
|
private readonly tbank: TbankClient,
|
|
private readonly logger: StructuredLogger,
|
|
private readonly loader: CatalogSourceLoader,
|
|
) {}
|
|
|
|
async readMany(
|
|
userId: string,
|
|
instrumentType: 'bond' | 'share',
|
|
uids: string[],
|
|
observe: boolean,
|
|
kinds: CatalogSource['kind'][] = ['last-price', 'detail'],
|
|
range?: {
|
|
from: string;
|
|
ticker?: string;
|
|
tickers?: Record<string, string>;
|
|
to: string;
|
|
},
|
|
refreshBefore?: Date,
|
|
) {
|
|
const { revision } = await this.access.local(userId);
|
|
const inputs = sourceInputs(instrumentType, uids, kinds, range);
|
|
const rows = inputs.length ? await this.repository.readSources(inputs) : [];
|
|
if (!observe) {
|
|
await this.requestMany(
|
|
userId,
|
|
revision,
|
|
instrumentType,
|
|
uids,
|
|
false,
|
|
kinds,
|
|
range,
|
|
refreshBefore,
|
|
);
|
|
}
|
|
const work = inputs.length
|
|
? await this.repository.workMany(userId, revision, inputs)
|
|
: [];
|
|
const byScope = new Map(work.map((row) => [row.scopeKey, row]));
|
|
return inputs.map((input) => {
|
|
const row = rows.find(
|
|
(entry) =>
|
|
entry.instrumentUid === input.uid && entry.kind === input.kind,
|
|
);
|
|
const payload = row?.payload as CatalogPayload | undefined;
|
|
return {
|
|
input,
|
|
version: row
|
|
? {
|
|
instrumentUid: row.instrumentUid,
|
|
source: row.source,
|
|
kind: row.kind,
|
|
rangeKey: row.rangeKey,
|
|
dataRevision: row.dataRevision,
|
|
fetchedAt: row.fetchedAt,
|
|
}
|
|
: null,
|
|
payload,
|
|
instrument: payload?.instrument ?? null,
|
|
price: payload?.price ?? null,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: catalogScope(input),
|
|
dataRevision: String(row?.dataRevision ?? MISSING_DATA_REVISION),
|
|
availability: row ? 'complete' : 'missing',
|
|
fetchedAt: row?.fetchedAt ?? null,
|
|
freshForMs: sourcePolicy(input.kind) === 'quote' ? tbankSourceTtlMs.quote : Number.MAX_SAFE_INTEGER,
|
|
work: byScope.get(catalogScope(input)) ?? null,
|
|
}),
|
|
};
|
|
});
|
|
}
|
|
|
|
requestMany(
|
|
userId: string,
|
|
revision: string,
|
|
instrumentType: 'bond' | 'share',
|
|
uids: string[],
|
|
force = false,
|
|
kinds: CatalogSource['kind'][] = ['last-price', 'detail'],
|
|
range?: {
|
|
from: string;
|
|
ticker?: string;
|
|
tickers?: Record<string, string>;
|
|
to: string;
|
|
},
|
|
refreshBefore?: Date,
|
|
) {
|
|
const inputs = sourceInputs(instrumentType, uids, kinds, range);
|
|
const reservation = this.reservations.then(async () => {
|
|
const rows = inputs.length
|
|
? await this.repository.readSources(inputs)
|
|
: [];
|
|
const ids: string[] = [];
|
|
for (const input of inputs) {
|
|
const row = rows.find(
|
|
(entry) =>
|
|
entry.instrumentUid === input.uid && entry.kind === input.kind,
|
|
);
|
|
const policy = sourcePolicy(input.kind);
|
|
const needsRefresh = force || !row?.fetchedAt ||
|
|
(policy === 'quote' && tbankSourceNeedsRefresh('quote', row.fetchedAt, new Date())) ||
|
|
// A walk reads a saved calendar once: later accounts see the new fetchedAt.
|
|
(policy === 'calendar' && refreshBefore !== undefined && row.fetchedAt < refreshBefore);
|
|
if (!needsRefresh) {
|
|
continue;
|
|
}
|
|
const reserved = await this.repository.reserve(
|
|
{ userId, revision },
|
|
input,
|
|
this.runId,
|
|
this.active,
|
|
);
|
|
if (!reserved) continue;
|
|
const { id, attemptId } = reserved.work;
|
|
ids.push(id);
|
|
if (!reserved.launch) continue;
|
|
const attempt = { userId, revision, id, attemptId };
|
|
this.active.add(attemptId);
|
|
let resolve = () => {};
|
|
const done = new Promise<void>((finish) => {
|
|
resolve = finish;
|
|
});
|
|
this.completions.set(attemptId, { attempt, done, resolve, workId: id });
|
|
const key = producerKey(input);
|
|
const producer = this.producers.get(key);
|
|
if (producer) producer.observers.push({ attempt, input });
|
|
else {
|
|
const queued = { input, attempt, observers: [] };
|
|
this.producers.set(key, queued);
|
|
this.queue.push(queued);
|
|
}
|
|
}
|
|
if (this.queue.length && !this.running) {
|
|
this.running = true;
|
|
setImmediate(() => {
|
|
void this.drain();
|
|
});
|
|
}
|
|
return ids;
|
|
});
|
|
this.reservations = reservation.catch(() => undefined);
|
|
return reservation;
|
|
}
|
|
|
|
async section(
|
|
userId: string,
|
|
instrumentType: 'bond' | 'share',
|
|
uid: string,
|
|
section: 'company' | 'metrics' | 'quote',
|
|
observe = false,
|
|
) {
|
|
if (section === 'company' && !observe) {
|
|
const { description, fundamentals, instrument } = await this.access.online(userId, async ({ token }) => {
|
|
const instrument =
|
|
instrumentType === 'share'
|
|
? await this.tbank.getShare(token, uid)
|
|
: await this.tbank.getBond(token, uid);
|
|
if (!instrument) throw new NotFoundException();
|
|
const [description, fundamentals] = instrument.assetUid
|
|
? await Promise.all([
|
|
this.tbank.getAssetDescription(token, instrument.assetUid),
|
|
this.tbank.getFundamentals(token, instrument.assetUid),
|
|
])
|
|
: [null, null];
|
|
return { description, fundamentals, instrument };
|
|
});
|
|
const meta = investmentResourceMeta({
|
|
resourceKey: `${instrumentType}/company/${uid}`,
|
|
dataRevision: `online:${Date.now()}`,
|
|
availability: 'complete',
|
|
fetchedAt: new Date(),
|
|
freshForMs: Number.MAX_SAFE_INTEGER,
|
|
work: null,
|
|
});
|
|
return {
|
|
values: { instrument: publicInstrument({ ...instrument, uid }), description, fundamentals },
|
|
meta,
|
|
unavailable: [],
|
|
observedAt: meta.lastSuccessfulAt ?? new Date().toISOString(),
|
|
};
|
|
}
|
|
const kinds: CatalogSource['kind'][] =
|
|
section === 'company'
|
|
? ['detail', 'description', 'fundamentals']
|
|
: [
|
|
...(instrumentType === 'bond'
|
|
? ['bond-current' as const]
|
|
: ['detail' as const]),
|
|
'last-price',
|
|
'trading-status',
|
|
'comparison',
|
|
...(section === 'metrics'
|
|
? ['coupons' as const, 'yield' as const, 'rating' as const]
|
|
: []),
|
|
];
|
|
const sources = await this.readMany(
|
|
userId,
|
|
instrumentType,
|
|
[uid],
|
|
observe,
|
|
kinds,
|
|
);
|
|
const first = sources[0];
|
|
if (first.meta.completeness === 'complete' && !first.instrument) {
|
|
throw new NotFoundException();
|
|
}
|
|
const values = sources.reduce<CatalogPayload>(
|
|
(result, source) => ({ ...result, ...source.payload }),
|
|
{},
|
|
);
|
|
const dependencies = sources.map((source) => source.meta);
|
|
const complete = dependencies.every(
|
|
(dependency) => dependency.completeness === 'complete',
|
|
);
|
|
const meta = investmentResourceMeta({
|
|
resourceKey: `${instrumentType}/${section}/${uid}`,
|
|
dataRevision: dependencies
|
|
.map((dependency) => dependency.dataRevision)
|
|
.join('/'),
|
|
availability: !first.instrument
|
|
? 'missing'
|
|
: complete
|
|
? 'complete'
|
|
: 'partial',
|
|
fetchedAt: first.meta.lastSuccessfulAt
|
|
? new Date(first.meta.lastSuccessfulAt)
|
|
: null,
|
|
freshForMs: section === 'company' ? Number.MAX_SAFE_INTEGER : tbankSourceTtlMs.quote,
|
|
work: null,
|
|
dependencies,
|
|
});
|
|
return {
|
|
values,
|
|
meta,
|
|
unavailable: sources
|
|
.filter(
|
|
(source) =>
|
|
source.meta.completeness !== 'complete' ||
|
|
source.meta.sync.state === 'failed',
|
|
)
|
|
.map((source) => source.input.kind),
|
|
observedAt: meta.lastSuccessfulAt ?? new Date().toISOString(),
|
|
};
|
|
}
|
|
|
|
// eslint-disable-next-line sonarjs/cognitive-complexity -- Queue draining coordinates publication fences and observer completion.
|
|
private async drain() {
|
|
try {
|
|
while (this.queue.length) {
|
|
const producer = this.queue.shift();
|
|
if (!producer) break;
|
|
const batch = [producer];
|
|
// One batch RPC for queued prices sharing the same authorized producer.
|
|
if (producer.input.kind === 'last-price') {
|
|
for (
|
|
let i = FIRST_QUEUE_INDEX;
|
|
i < this.queue.length && batch.length < catalogSearchLimit;
|
|
) {
|
|
// eslint-disable-next-line security/detect-object-injection -- Numeric queue indexes and validated instrument UIDs select internal data only.
|
|
const next = this.queue[i];
|
|
if (
|
|
next.input.kind === 'last-price' &&
|
|
next.attempt.userId === producer.attempt.userId &&
|
|
next.attempt.revision === producer.attempt.revision
|
|
) {
|
|
batch.push(next);
|
|
this.queue.splice(i, DEQUEUED_ITEM_COUNT);
|
|
} else i++;
|
|
}
|
|
}
|
|
let prices: TbankLastPrice[] | undefined;
|
|
let payload: CatalogPayload | undefined;
|
|
let failed = false;
|
|
try {
|
|
const token = await this.access.tokenForAttempt(producer.attempt);
|
|
if (!token) failed = true;
|
|
else if (producer.input.kind === 'last-price') {
|
|
prices = await this.tbank.getLastPrices(
|
|
token,
|
|
batch.map((item) => item.input.uid),
|
|
);
|
|
} else {
|
|
payload = await this.loader.load(producer.input, token, producer.attempt);
|
|
}
|
|
} catch (error) {
|
|
failed = true;
|
|
if (error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)) {
|
|
await this.access.rejectMembershipToken(producer.attempt);
|
|
} else {
|
|
this.logger.event('warn', 'catalog-enrichment.failed', {
|
|
code: TbankErrorCode.Unavailable,
|
|
});
|
|
}
|
|
}
|
|
for (const item of batch) {
|
|
const price =
|
|
prices?.find((entry) => entry.instrumentUid === item.input.uid) ??
|
|
null;
|
|
const value =
|
|
item.input.kind === 'last-price'
|
|
? {
|
|
price: price
|
|
? {
|
|
instrumentUid: item.input.uid,
|
|
price: price.price,
|
|
time: price.time,
|
|
}
|
|
: null,
|
|
}
|
|
: payload;
|
|
// Keep accepting observers while publication awaits SQLite.
|
|
item.observers.unshift({ attempt: item.attempt, input: item.input });
|
|
for (const observer of item.observers) {
|
|
const success =
|
|
!failed &&
|
|
value !== undefined &&
|
|
(await this.repository.publishSource(
|
|
observer.attempt,
|
|
observer.input,
|
|
value,
|
|
new Date(),
|
|
producer.attempt,
|
|
));
|
|
await this.repository.finish(observer.attempt, success);
|
|
this.active.delete(observer.attempt.attemptId);
|
|
this.completions.get(observer.attempt.attemptId)?.resolve();
|
|
this.completions.delete(observer.attempt.attemptId);
|
|
}
|
|
this.producers.delete(producerKey(item.input));
|
|
}
|
|
}
|
|
} catch {
|
|
// Persistence failed mid-batch: finish every still-open observer
|
|
// attempt durably so no source row is left running. Best effort — a
|
|
// row that stays open is re-reserved by the next reader.
|
|
for (const completion of this.completions.values()) {
|
|
await this.repository
|
|
.finish(completion.attempt, false)
|
|
.catch(() => undefined);
|
|
}
|
|
for (const completion of this.completions.values()) completion.resolve();
|
|
this.completions.clear();
|
|
this.active.clear();
|
|
this.producers.clear();
|
|
this.queue.length = 0;
|
|
this.logger.event('error', 'catalog-enrichment.persistence-failed', {
|
|
code: TbankErrorCode.Unavailable,
|
|
});
|
|
} finally {
|
|
this.running = false;
|
|
}
|
|
}
|
|
}
|
|
function sourceInputs(
|
|
instrumentType: 'bond' | 'share',
|
|
uids: string[],
|
|
kinds: CatalogSource['kind'][],
|
|
range?: {
|
|
from: string;
|
|
ticker?: string;
|
|
tickers?: Record<string, string>;
|
|
to: string;
|
|
},
|
|
): CatalogSource[] {
|
|
return kinds.flatMap((kind) =>
|
|
[...new Set(uids)].map((uid) => ({
|
|
instrumentType,
|
|
uid,
|
|
kind,
|
|
// eslint-disable-next-line security/detect-object-injection -- Numeric queue indexes and validated instrument UIDs select internal data only.
|
|
...(kind === 'bond-events' && (range?.tickers?.[uid] || range?.ticker)
|
|
// eslint-disable-next-line security/detect-object-injection -- Numeric queue indexes and validated instrument UIDs select internal data only.
|
|
? { ticker: range?.tickers?.[uid] || range?.ticker }
|
|
: {}),
|
|
...(['coupons', 'dividends'].includes(kind)
|
|
? {
|
|
rangeKey: range
|
|
? JSON.stringify([range.from, range.to])
|
|
: String(new Date().getUTCFullYear()),
|
|
}
|
|
: {}),
|
|
})),
|
|
);
|
|
}
|
|
|
|
function sourcePolicy(kind: CatalogSource['kind']) {
|
|
if (
|
|
[
|
|
'bond-current',
|
|
'comparison',
|
|
'last-price',
|
|
'trading-status',
|
|
'yield',
|
|
].includes(kind)
|
|
) {
|
|
return 'quote';
|
|
}
|
|
if (['bond-events', 'dividends'].includes(kind)) return 'calendar';
|
|
return 'stable';
|
|
}
|
|
function producerKey(input: CatalogSource) {
|
|
return input.instrumentType === 'bond' &&
|
|
(input.kind === 'detail' || input.kind === 'bond-current')
|
|
? `bond-metadata:${input.uid}`
|
|
: catalogScope(input);
|
|
}
|