532 lines
17 KiB
TypeScript
532 lines
17 KiB
TypeScript
import {
|
|
ConflictException,
|
|
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 {
|
|
TbankAccessService,
|
|
type TbankBond,
|
|
TbankClient,
|
|
type TbankShare,
|
|
tbankSourceTtlMs,
|
|
type TbankSyncAttempt,
|
|
} from '../../../../integrations/tbank/tbank.api';
|
|
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
|
import { InstrumentErrorCode } from '../domain/instrument-error-code';
|
|
import {
|
|
CatalogDiscoveryRepository,
|
|
catalogQuery,
|
|
type CatalogQuery,
|
|
catalogScope,
|
|
} from '../infrastructure/catalog-discovery.repository';
|
|
import { CatalogEnrichmentService } from './catalog-enrichment.service';
|
|
import {
|
|
catalogSearchLimit,
|
|
selectCatalogInstruments,
|
|
} from './catalog-selection';
|
|
|
|
const UNAMBIGUOUS_MATCH_COUNT = 1;
|
|
const NO_MATCHES = 0;
|
|
const FIRST_RESULT_OFFSET = 0;
|
|
const MISSING_DATA_REVISION = '0';
|
|
|
|
const tickerAmbiguityLimit = 2;
|
|
|
|
type Producer = {
|
|
attempt: TbankSyncAttempt;
|
|
force: boolean;
|
|
input: CatalogQuery;
|
|
observers: TbankSyncAttempt[];
|
|
};
|
|
@Injectable()
|
|
export class CatalogDiscoveryService {
|
|
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,
|
|
{ done: Promise<void>; resolve: () => void; workId: string }
|
|
>();
|
|
|
|
constructor(
|
|
private readonly repository: CatalogDiscoveryRepository,
|
|
private readonly access: TbankAccessService,
|
|
private readonly tbank: TbankClient,
|
|
private readonly logger: StructuredLogger,
|
|
private readonly enrichment: CatalogEnrichmentService,
|
|
) {}
|
|
|
|
async read(userId: string, input: CatalogQuery, observe = false) {
|
|
input = catalogQuery(input);
|
|
const connection = await this.access.local(userId);
|
|
const saved = await this.repository.read(input);
|
|
if (!observe) await this.request(userId, connection.revision, input);
|
|
return {
|
|
data: saved.data,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: catalogScope(input),
|
|
dataRevision: saved.dataRevision,
|
|
availability: saved.data === null ? 'missing' : 'complete',
|
|
fetchedAt: saved.fetchedAt,
|
|
freshForMs: Number.MAX_SAFE_INTEGER,
|
|
work: await this.repository.work(userId, connection.revision, input),
|
|
}),
|
|
};
|
|
}
|
|
|
|
async search(userId: string, input: CatalogQuery, observe = false) {
|
|
if (!observe) return this.onlineSearch(userId, catalogQuery(input));
|
|
const discovery = await this.read(userId, input, observe);
|
|
const sources = discovery.data
|
|
? await this.enrichment.readMany(
|
|
userId,
|
|
input.instrumentType,
|
|
discovery.data.map((item) => item.uid),
|
|
observe,
|
|
)
|
|
: [];
|
|
const dependencies = [
|
|
discovery.meta,
|
|
...sources.map((source) => source.meta),
|
|
];
|
|
return {
|
|
data:
|
|
discovery.data?.map((short) => ({
|
|
...short,
|
|
detail:
|
|
sources.find(
|
|
(source) =>
|
|
source.input.uid === short.uid &&
|
|
source.input.kind === 'detail',
|
|
)?.instrument ?? null,
|
|
price:
|
|
sources.find(
|
|
(source) =>
|
|
source.input.uid === short.uid &&
|
|
source.input.kind === 'last-price',
|
|
)?.price ?? null,
|
|
})) ?? null,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: `search:${catalogScope(catalogQuery(input))}`,
|
|
dataRevision: dependencies
|
|
.map((dependency) => dependency.dataRevision)
|
|
.join('/'),
|
|
availability:
|
|
discovery.data === null
|
|
? 'missing'
|
|
: sources.every((source) => source.meta.completeness === 'complete')
|
|
? 'complete'
|
|
: 'partial',
|
|
fetchedAt: discovery.meta.lastSuccessfulAt
|
|
? new Date(discovery.meta.lastSuccessfulAt)
|
|
: null,
|
|
freshForMs: Number.MAX_SAFE_INTEGER,
|
|
work: discovery.meta.sync.id
|
|
? {
|
|
id: discovery.meta.sync.id,
|
|
state: discovery.meta.sync.state,
|
|
errorCode: discovery.meta.sync.errorCode,
|
|
}
|
|
: null,
|
|
dependencies,
|
|
}),
|
|
};
|
|
}
|
|
|
|
async detail(
|
|
userId: string,
|
|
instrumentType: 'bond' | 'share',
|
|
uid: string,
|
|
observe = false,
|
|
) {
|
|
if (!observe) {
|
|
const token = await this.access.onlineToken(userId);
|
|
const instrument =
|
|
instrumentType === 'share'
|
|
? await this.tbank.getShare(token, uid)
|
|
: await this.tbank.getBond(token, uid);
|
|
if (!instrument) throw new NotFoundException();
|
|
return {
|
|
data: { ...publicInstrument({ ...instrument, uid }), uid, slug: instrument.ticker?.toUpperCase() ?? uid },
|
|
meta: investmentResourceMeta({
|
|
resourceKey: `${instrumentType}/detail/${uid}`,
|
|
dataRevision: `online:${Date.now()}`,
|
|
availability: 'complete',
|
|
fetchedAt: new Date(),
|
|
freshForMs: tbankSourceTtlMs.portfolio,
|
|
work: null,
|
|
}),
|
|
};
|
|
}
|
|
const sources = await this.enrichment.readMany(
|
|
userId,
|
|
instrumentType,
|
|
[uid],
|
|
observe,
|
|
['detail'],
|
|
);
|
|
const source = sources[0];
|
|
if (source.meta.completeness === 'complete' && !source.instrument) {
|
|
throw new NotFoundException();
|
|
}
|
|
const instrument = source.instrument;
|
|
let slug = uid;
|
|
const dependencies = [source.meta];
|
|
if (
|
|
instrument?.ticker &&
|
|
/^[A-Z0-9][\w.@+-]{0,79}$/i.test(instrument.ticker)
|
|
) {
|
|
const alias = await this.read(
|
|
userId,
|
|
{ instrumentType, query: instrument.ticker, exactTicker: true },
|
|
observe,
|
|
);
|
|
if (alias.data?.length === UNAMBIGUOUS_MATCH_COUNT && alias.data[0].uid === uid) {
|
|
slug = instrument.ticker.toUpperCase();
|
|
}
|
|
dependencies.push(alias.meta);
|
|
}
|
|
return {
|
|
data: instrument ? { ...instrument, uid, slug } : null,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: `${instrumentType}/detail/${uid}`,
|
|
dataRevision: dependencies
|
|
.map((dependency) => dependency.dataRevision)
|
|
.join('/'),
|
|
availability: source.meta.availability,
|
|
fetchedAt: source.meta.lastSuccessfulAt
|
|
? new Date(source.meta.lastSuccessfulAt)
|
|
: null,
|
|
freshForMs: tbankSourceTtlMs.portfolio,
|
|
work: source.meta.sync.id
|
|
? {
|
|
id: source.meta.sync.id,
|
|
state: source.meta.sync.state,
|
|
errorCode: source.meta.sync.errorCode,
|
|
}
|
|
: null,
|
|
dependencies,
|
|
}),
|
|
};
|
|
}
|
|
|
|
// eslint-disable-next-line sonarjs/cognitive-complexity -- Online and observe paths intentionally preserve distinct error semantics.
|
|
async byTicker(
|
|
userId: string,
|
|
instrumentType: 'bond' | 'share',
|
|
ticker: string,
|
|
observe = false,
|
|
) {
|
|
if (!observe) {
|
|
const query = ticker.trim().toUpperCase();
|
|
const token = await this.access.onlineToken(userId);
|
|
const found =
|
|
instrumentType === 'share'
|
|
? await this.tbank.findShares(token, query)
|
|
: await this.tbank.findBonds(token, query);
|
|
const matches = selectCatalogInstruments(
|
|
found.filter((item) => item.ticker?.trim().toUpperCase() === query),
|
|
query,
|
|
instrumentType,
|
|
).slice(FIRST_RESULT_OFFSET, tickerAmbiguityLimit);
|
|
if (!matches.length) throw new NotFoundException();
|
|
if (matches.length > UNAMBIGUOUS_MATCH_COUNT) {
|
|
throw new ConflictException({
|
|
code:
|
|
instrumentType === 'share'
|
|
? InstrumentErrorCode.StockTickerAmbiguous
|
|
: InstrumentErrorCode.BondTickerAmbiguous,
|
|
});
|
|
}
|
|
const result = await this.detail(userId, instrumentType, matches[0].uid);
|
|
return { ...result, data: result.data ? { ...result.data, slug: query } : null };
|
|
}
|
|
const lookup = await this.read(
|
|
userId,
|
|
{ instrumentType, query: ticker, exactTicker: true },
|
|
observe,
|
|
);
|
|
if (lookup.data?.length === NO_MATCHES) throw new NotFoundException();
|
|
if (lookup.data && lookup.data.length > UNAMBIGUOUS_MATCH_COUNT) {
|
|
throw new ConflictException({
|
|
code:
|
|
instrumentType === 'share'
|
|
? InstrumentErrorCode.StockTickerAmbiguous
|
|
: InstrumentErrorCode.BondTickerAmbiguous,
|
|
});
|
|
}
|
|
const result = lookup.data
|
|
? await this.detail(userId, instrumentType, lookup.data[0].uid, observe)
|
|
: null;
|
|
const dependencies = [lookup.meta, ...(result ? [result.meta] : [])];
|
|
return {
|
|
data: result?.data
|
|
? { ...result.data, slug: ticker.trim().toUpperCase() }
|
|
: null,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: `${instrumentType}/ticker/${ticker.trim().toUpperCase()}`,
|
|
dataRevision: dependencies
|
|
.map((dependency) => dependency.dataRevision)
|
|
.join('/'),
|
|
availability: result?.meta.availability ?? 'missing',
|
|
fetchedAt: lookup.meta.lastSuccessfulAt
|
|
? new Date(lookup.meta.lastSuccessfulAt)
|
|
: null,
|
|
freshForMs: tbankSourceTtlMs.portfolio,
|
|
work: lookup.meta.sync.id
|
|
? {
|
|
id: lookup.meta.sync.id,
|
|
state: lookup.meta.sync.state,
|
|
errorCode: lookup.meta.sync.errorCode,
|
|
}
|
|
: null,
|
|
dependencies,
|
|
}),
|
|
};
|
|
}
|
|
|
|
private async onlineSearch(userId: string, input: CatalogQuery) {
|
|
const token = await this.access.onlineToken(userId);
|
|
const found =
|
|
input.instrumentType === 'share'
|
|
? await this.tbank.findShares(token, input.query)
|
|
: await this.tbank.findBonds(token, input.query);
|
|
const instruments = selectCatalogInstruments(
|
|
input.exactTicker
|
|
? found.filter((item) => item.ticker?.trim().toUpperCase() === input.query)
|
|
: found,
|
|
input.query,
|
|
input.instrumentType,
|
|
).slice(FIRST_RESULT_OFFSET, input.exactTicker ? tickerAmbiguityLimit : catalogSearchLimit);
|
|
const details = await Promise.all(
|
|
instruments.map((item) =>
|
|
input.instrumentType === 'share'
|
|
? this.tbank.getShare(token, item.uid)
|
|
: this.tbank.getBond(token, item.uid),
|
|
),
|
|
);
|
|
const prices = instruments.length
|
|
? await this.tbank.getLastPrices(token, instruments.map((item) => item.uid))
|
|
: [];
|
|
const fetchedAt = new Date();
|
|
const data = instruments.flatMap((item, index) => {
|
|
const detail = details.at(index);
|
|
return [{
|
|
...item,
|
|
detail: detail ? publicInstrument({ ...detail, uid: item.uid }) : null,
|
|
price: prices.find((price) => price.instrumentUid === item.uid) ?? null,
|
|
}];
|
|
});
|
|
return {
|
|
data,
|
|
meta: investmentResourceMeta({
|
|
resourceKey: `search:${catalogScope(input)}`,
|
|
dataRevision: `online:${fetchedAt.toISOString()}`,
|
|
availability: 'complete',
|
|
fetchedAt,
|
|
freshForMs: tbankSourceTtlMs.portfolio,
|
|
work: null,
|
|
}),
|
|
};
|
|
}
|
|
|
|
request(
|
|
userId: string,
|
|
revision: string,
|
|
query: CatalogQuery,
|
|
force = false,
|
|
) {
|
|
const input = catalogQuery(query);
|
|
const reserved = this.reservations.then(async () => {
|
|
const saved = await this.repository.read(input);
|
|
if (
|
|
!force &&
|
|
saved.dataRevision !== MISSING_DATA_REVISION
|
|
) return null;
|
|
const reservation = await this.repository.reserve(
|
|
{ userId, revision },
|
|
input,
|
|
this.runId,
|
|
this.active,
|
|
);
|
|
if (!reservation) return null;
|
|
const { id, attemptId } = reservation.work;
|
|
if (!reservation.launch) {
|
|
const current = this.producers.get(catalogScope(input));
|
|
if (current) current.force ||= force;
|
|
return id;
|
|
}
|
|
const attempt = { userId, revision, id, attemptId };
|
|
this.active.add(attemptId);
|
|
let resolve = () => {};
|
|
const done = new Promise<void>((finish) => {
|
|
resolve = finish;
|
|
});
|
|
this.completions.set(attemptId, { workId: id, done, resolve });
|
|
const key = catalogScope(input);
|
|
const producer = this.producers.get(key);
|
|
if (producer) {
|
|
producer.observers.push(attempt);
|
|
producer.force ||= force;
|
|
} else {
|
|
const queued = { input, attempt, observers: [], force };
|
|
this.producers.set(key, queued);
|
|
this.queue.push(queued);
|
|
}
|
|
if (!this.running) {
|
|
this.running = true;
|
|
setImmediate(() => {
|
|
void this.drain();
|
|
});
|
|
}
|
|
return id;
|
|
});
|
|
this.reservations = reserved.catch(() => undefined);
|
|
return reserved;
|
|
}
|
|
|
|
async refresh(userId: string, revision: string, query: CatalogQuery) {
|
|
const id = await this.request(userId, revision, query, true);
|
|
if (id) {
|
|
await Promise.all(
|
|
[...this.completions.values()]
|
|
.filter((entry) => entry.workId === id)
|
|
.map((entry) => entry.done),
|
|
);
|
|
}
|
|
const saved = await this.repository.read(catalogQuery(query));
|
|
const sources = await this.enrichment.readMany(
|
|
userId,
|
|
query.instrumentType,
|
|
saved.data?.map((item) => item.uid) ?? [],
|
|
true,
|
|
);
|
|
return [
|
|
...(id ? [id] : []),
|
|
...sources.flatMap((source) =>
|
|
source.meta.sync.id ? [source.meta.sync.id] : [],
|
|
),
|
|
];
|
|
}
|
|
|
|
private async drain() {
|
|
try {
|
|
while (this.queue.length) {
|
|
const producer = this.queue.shift();
|
|
if (!producer) break;
|
|
const { input, attempt } = producer;
|
|
let success = false;
|
|
try {
|
|
const token = await this.access.tokenForAttempt(attempt);
|
|
if (token) {
|
|
const found =
|
|
input.instrumentType === 'share'
|
|
? await this.tbank.findShares(token, input.query)
|
|
: await this.tbank.findBonds(token, input.query);
|
|
success = await this.repository.publish(
|
|
attempt,
|
|
input,
|
|
selectCatalogInstruments(
|
|
input.exactTicker
|
|
? found.filter(
|
|
(item) =>
|
|
item.ticker?.trim().toUpperCase() === input.query,
|
|
)
|
|
: found,
|
|
input.query,
|
|
input.instrumentType,
|
|
).slice(
|
|
FIRST_RESULT_OFFSET,
|
|
input.exactTicker ? tickerAmbiguityLimit : catalogSearchLimit,
|
|
),
|
|
new Date(),
|
|
);
|
|
}
|
|
} catch (error) {
|
|
if (error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)) {
|
|
await this.access.rejectMembershipToken(attempt);
|
|
} else {
|
|
this.logger.event('warn', 'catalog-discovery.failed', {
|
|
code: TbankErrorCode.Unavailable,
|
|
});
|
|
}
|
|
}
|
|
if (success) {
|
|
const saved = await this.repository.read(input);
|
|
for (const observer of [attempt, ...producer.observers]) {
|
|
if (!input.exactTicker || saved.data?.length === UNAMBIGUOUS_MATCH_COUNT) {
|
|
await this.enrichment.requestMany(
|
|
observer.userId,
|
|
observer.revision,
|
|
input.instrumentType,
|
|
saved.data?.map((item) => item.uid) ?? [],
|
|
producer.force,
|
|
input.exactTicker ? ['detail'] : ['last-price', 'detail'],
|
|
);
|
|
}
|
|
}
|
|
}
|
|
this.producers.delete(catalogScope(input));
|
|
for (const observer of [attempt, ...producer.observers]) {
|
|
await this.repository.finish(observer, success);
|
|
this.active.delete(observer.attemptId);
|
|
this.completions.get(observer.attemptId)?.resolve();
|
|
this.completions.delete(observer.attemptId);
|
|
}
|
|
}
|
|
} catch {
|
|
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-discovery.persistence-failed', {
|
|
code: TbankErrorCode.Unavailable,
|
|
});
|
|
} finally {
|
|
this.running = false;
|
|
}
|
|
}
|
|
}
|
|
|
|
function publicInstrument(value: TbankBond | TbankShare) {
|
|
const bond = value as TbankBond;
|
|
const share = value as TbankShare;
|
|
return {
|
|
uid: value.uid,
|
|
figi: value.figi,
|
|
assetUid: value.assetUid,
|
|
name: value.name,
|
|
ticker: value.ticker,
|
|
isin: value.isin,
|
|
classCode: value.classCode,
|
|
realExchange: value.realExchange,
|
|
apiTradeAvailableFlag: value.apiTradeAvailableFlag,
|
|
currency: value.currency,
|
|
shareType: share.shareType,
|
|
sector: value.sector,
|
|
lot: value.lot,
|
|
brand: value.brand ? { logoName: value.brand.logoName } : undefined,
|
|
first_1dayCandleDate: value.first_1dayCandleDate,
|
|
nominal: bond.nominal,
|
|
initialNominal: bond.initialNominal,
|
|
aciValue: bond.aciValue,
|
|
maturityDate: bond.maturityDate,
|
|
callDate: bond.callDate,
|
|
couponQuantityPerYear: bond.couponQuantityPerYear,
|
|
floatingCouponFlag: bond.floatingCouponFlag,
|
|
perpetualFlag: bond.perpetualFlag,
|
|
amortizationFlag: bond.amortizationFlag,
|
|
subordinatedFlag: bond.subordinatedFlag,
|
|
bondType: bond.bondType,
|
|
riskLevel: bond.riskLevel,
|
|
};
|
|
}
|