494 lines
16 KiB
TypeScript
494 lines
16 KiB
TypeScript
import { Injectable } from '@nestjs/common';
|
|
import { randomUUID } from 'node:crypto';
|
|
import { isDeepStrictEqual } from 'node:util';
|
|
|
|
import type { Prisma } from '../../../../../generated/prisma/client';
|
|
|
|
import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
|
|
import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
|
|
import { PrismaService } from '../../../../prisma/prisma.service';
|
|
|
|
const UNCHANGED_REVISION_INCREMENT = 0;
|
|
const REVISION_INCREMENT = 1;
|
|
|
|
export type CatalogQuery = {
|
|
exactTicker?: boolean;
|
|
instrumentType: 'bond' | 'share';
|
|
query: string;
|
|
};
|
|
export type CatalogSource = {
|
|
instrumentType: 'bond' | 'share';
|
|
kind:
|
|
| 'bond-current' |
|
|
'bond-events' |
|
|
'comparison' |
|
|
'coupons' |
|
|
'description' |
|
|
'detail' |
|
|
'dividends' |
|
|
'fundamentals' |
|
|
'last-price' |
|
|
'rating' |
|
|
'trading-status' |
|
|
'yield';
|
|
rangeKey?: string;
|
|
ticker?: string;
|
|
uid: string;
|
|
};
|
|
export type CatalogWork = CatalogQuery | CatalogSource;
|
|
export type CatalogInstrument = (TbankBond | TbankShare) & { uid: string };
|
|
const provider = 'tbank';
|
|
export function catalogQuery(input: CatalogQuery): CatalogQuery {
|
|
return {
|
|
instrumentType: input.instrumentType,
|
|
query: input.query.trim().toUpperCase(),
|
|
...(input.exactTicker ? { exactTicker: true } : {}),
|
|
};
|
|
}
|
|
export function catalogScope(input: CatalogWork) {
|
|
if ('uid' in input) {
|
|
return `catalog-source:${JSON.stringify([
|
|
input.kind,
|
|
input.kind === 'detail' ? input.instrumentType : null,
|
|
input.rangeKey ?? '',
|
|
input.uid,
|
|
])}`;
|
|
}
|
|
return `${input.exactTicker ? 'ticker' : 'catalog'}:${JSON.stringify([input.instrumentType, input.query])}`;
|
|
}
|
|
|
|
@Injectable()
|
|
export class CatalogDiscoveryRepository {
|
|
constructor(private readonly prisma: PrismaService) {}
|
|
|
|
read(input: CatalogQuery) {
|
|
return this.prisma.$transaction(async (tx) => {
|
|
const alias = input.exactTicker
|
|
? await tx.tbankInstrumentLookup.findUnique({
|
|
where: {
|
|
provider_identifierType_identifier: {
|
|
provider,
|
|
identifierType: `${input.instrumentType}-ticker`,
|
|
identifier: input.query,
|
|
},
|
|
},
|
|
})
|
|
: null;
|
|
const discovery = input.exactTicker
|
|
? alias
|
|
? {
|
|
instrumentUids: alias.matchedUids,
|
|
fetchedAt: alias.fetchedAt,
|
|
dataRevision: alias.dataRevision,
|
|
}
|
|
: null
|
|
: await tx.tbankCatalogDiscovery.findUnique({
|
|
where: {
|
|
provider_instrumentType_query: {
|
|
provider,
|
|
instrumentType: input.instrumentType,
|
|
query: input.query,
|
|
},
|
|
},
|
|
});
|
|
if (!discovery) return { data: null, fetchedAt: null, dataRevision: '0' };
|
|
const uids = discovery.instrumentUids as string[];
|
|
const rows = await tx.tbankInstrumentSource.findMany({
|
|
where: {
|
|
provider,
|
|
instrumentUid: { in: uids },
|
|
source: provider,
|
|
kind: 'catalog',
|
|
rangeKey: input.instrumentType,
|
|
},
|
|
});
|
|
const byUid = new Map(rows.map((row) => [row.instrumentUid, row]));
|
|
// A damaged/missing dependency is unknown, never a confirmed empty search.
|
|
if (uids.some((uid) => !byUid.has(uid))) {
|
|
return {
|
|
data: null,
|
|
fetchedAt: null,
|
|
dataRevision: String(discovery.dataRevision),
|
|
};
|
|
}
|
|
return {
|
|
data: uids.map((uid) => byUid.get(uid)?.payload as CatalogInstrument),
|
|
fetchedAt: discovery.fetchedAt,
|
|
dataRevision: [
|
|
discovery.dataRevision,
|
|
...uids.map((uid) => byUid.get(uid)?.dataRevision),
|
|
].join('/'),
|
|
};
|
|
});
|
|
}
|
|
|
|
readSources(inputs: CatalogSource[]) {
|
|
// Share identity predicates across UIDs instead of preparing one OR per row.
|
|
const groups = new Map<string, CatalogSource[]>();
|
|
for (const input of inputs) {
|
|
const key = JSON.stringify([
|
|
input.kind,
|
|
sourceProvider(input),
|
|
input.kind === 'detail' ? input.instrumentType : (input.rangeKey ?? ''),
|
|
]);
|
|
const group = groups.get(key);
|
|
if (group) group.push(input);
|
|
else groups.set(key, [input]);
|
|
}
|
|
return this.prisma.tbankInstrumentSource.findMany({
|
|
where: {
|
|
provider,
|
|
OR: [...groups.values()].map(([input, ...rest]) => ({
|
|
instrumentUid: { in: [input, ...rest].map((entry) => entry.uid) },
|
|
kind: input.kind,
|
|
source: sourceProvider(input),
|
|
rangeKey:
|
|
input.kind === 'detail'
|
|
? input.instrumentType
|
|
: (input.rangeKey ?? ''),
|
|
})),
|
|
},
|
|
});
|
|
}
|
|
|
|
publishSource(
|
|
attempt: TbankSyncAttempt,
|
|
input: CatalogSource,
|
|
value: object,
|
|
now: Date,
|
|
initiator: TbankSyncAttempt = attempt,
|
|
) {
|
|
return this.prisma.$transaction(async (tx) => {
|
|
const origin = await tx.tbankSyncWork.updateMany({
|
|
where: {
|
|
...initiator,
|
|
state: { in: ['running', 'complete'] },
|
|
connection: { invalidToken: false },
|
|
},
|
|
data: { progress: {} },
|
|
});
|
|
if (!origin.count) return false;
|
|
const claimed = await tx.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: false },
|
|
},
|
|
data: { state: 'complete', progress: {} },
|
|
});
|
|
if (!claimed.count) return false;
|
|
const payload = JSON.parse(
|
|
JSON.stringify(value),
|
|
) as Prisma.InputJsonObject;
|
|
const identity = { provider, uid: input.uid };
|
|
await tx.tbankInstrument.upsert({
|
|
where: { provider_uid: identity },
|
|
update: {},
|
|
create: {
|
|
...identity,
|
|
metadata: {},
|
|
instrumentType: input.instrumentType,
|
|
fetchedAt: now,
|
|
},
|
|
});
|
|
const instrument = (
|
|
value as { instrument?: null | TbankBond | TbankShare }
|
|
).instrument;
|
|
if (
|
|
(input.kind === 'detail' || input.kind === 'bond-current') &&
|
|
instrument &&
|
|
(instrument.name?.trim() || instrument.ticker?.trim())
|
|
) {
|
|
const previous = await tx.tbankInstrument.findUniqueOrThrow({
|
|
where: { provider_uid: identity },
|
|
});
|
|
const metadata = JSON.parse(
|
|
JSON.stringify({
|
|
...instrument,
|
|
instrumentType: input.instrumentType,
|
|
}),
|
|
) as Prisma.InputJsonObject;
|
|
await tx.tbankInstrument.update({
|
|
where: { provider_uid: identity },
|
|
data: {
|
|
metadata,
|
|
instrumentType: input.instrumentType,
|
|
figi: instrument.figi ?? null,
|
|
ticker: instrument.ticker ?? null,
|
|
fetchedAt: now,
|
|
dataRevision: {
|
|
increment: isDeepStrictEqual(previous.metadata, metadata) ? UNCHANGED_REVISION_INCREMENT : REVISION_INCREMENT,
|
|
},
|
|
},
|
|
});
|
|
for (const [identifierType, identifier] of [
|
|
['uid', input.uid],
|
|
['figi', instrument.figi],
|
|
] as const) {
|
|
if (!identifier) continue;
|
|
const alias = { provider, identifierType, identifier };
|
|
const previousAlias = await tx.tbankInstrumentLookup.findUnique({
|
|
where: { provider_identifierType_identifier: alias },
|
|
});
|
|
const matchedUids = [input.uid];
|
|
await tx.tbankInstrumentLookup.upsert({
|
|
where: { provider_identifierType_identifier: alias },
|
|
create: { ...alias, matchedUids, fetchedAt: now },
|
|
update: {
|
|
matchedUids,
|
|
fetchedAt: now,
|
|
dataRevision: {
|
|
increment: isDeepStrictEqual(
|
|
previousAlias?.matchedUids,
|
|
matchedUids,
|
|
)
|
|
? UNCHANGED_REVISION_INCREMENT
|
|
: REVISION_INCREMENT,
|
|
},
|
|
},
|
|
});
|
|
}
|
|
}
|
|
const key = {
|
|
provider,
|
|
instrumentUid: input.uid,
|
|
source: sourceProvider(input),
|
|
kind: input.kind,
|
|
rangeKey:
|
|
input.kind === 'detail'
|
|
? input.instrumentType
|
|
: (input.rangeKey ?? ''),
|
|
};
|
|
const previous = await tx.tbankInstrumentSource.findUnique({
|
|
where: { provider_instrumentUid_source_kind_rangeKey: key },
|
|
});
|
|
await tx.tbankInstrumentSource.upsert({
|
|
where: { provider_instrumentUid_source_kind_rangeKey: key },
|
|
create: { ...key, payload, fetchedAt: now },
|
|
update: {
|
|
payload,
|
|
fetchedAt: now,
|
|
dataRevision: {
|
|
increment: isDeepStrictEqual(previous?.payload, payload) ? UNCHANGED_REVISION_INCREMENT : REVISION_INCREMENT,
|
|
},
|
|
},
|
|
});
|
|
return true;
|
|
});
|
|
}
|
|
|
|
reserve(
|
|
owner: { revision: string; userId: string },
|
|
input: CatalogWork,
|
|
runId: string,
|
|
active: ReadonlySet<string>,
|
|
) {
|
|
return this.prisma.$transaction(async (tx) => {
|
|
const current = await tx.tbankConnection.updateMany({
|
|
where: { ...owner, invalidToken: false },
|
|
data: { membershipRevision: { increment: 0 } },
|
|
});
|
|
if (!current.count) return null;
|
|
const scope = { ...owner, scopeKey: catalogScope(input) };
|
|
const previous = await tx.tbankSyncWork.findUnique({
|
|
where: { userId_revision_scopeKey: scope },
|
|
});
|
|
if (isJoinableRunningAttempt(previous, runId, active)) {
|
|
return { work: previous, launch: false };
|
|
}
|
|
const data = {
|
|
attemptId: randomUUID(),
|
|
processRunId: runId,
|
|
state: 'running',
|
|
priority: 1,
|
|
intent: { source: 'catalog', ...input },
|
|
progress: {},
|
|
errorCode: null,
|
|
};
|
|
const work = await tx.tbankSyncWork.upsert({
|
|
where: { userId_revision_scopeKey: scope },
|
|
create: { ...scope, ...data, id: randomUUID() },
|
|
update: data,
|
|
});
|
|
return { work, launch: true };
|
|
});
|
|
}
|
|
|
|
workMany(userId: string, revision: string, inputs: CatalogWork[]) {
|
|
return this.prisma.tbankSyncWork.findMany({
|
|
where: { userId, revision, scopeKey: { in: inputs.map(catalogScope) } },
|
|
select: { scopeKey: true, id: true, state: true, errorCode: true },
|
|
});
|
|
}
|
|
|
|
work(userId: string, revision: string, input: CatalogWork) {
|
|
return this.prisma.tbankSyncWork.findUnique({
|
|
where: {
|
|
userId_revision_scopeKey: {
|
|
userId,
|
|
revision,
|
|
scopeKey: catalogScope(input),
|
|
},
|
|
},
|
|
select: { id: true, state: true, errorCode: true },
|
|
});
|
|
}
|
|
|
|
publish(
|
|
attempt: TbankSyncAttempt,
|
|
input: CatalogQuery,
|
|
instruments: CatalogInstrument[],
|
|
now: Date,
|
|
) {
|
|
return this.prisma.$transaction(async (tx) => {
|
|
const claimed = await tx.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: false },
|
|
},
|
|
data: { state: 'complete', progress: {} },
|
|
});
|
|
if (!claimed.count) return false;
|
|
for (const instrument of instruments) {
|
|
// Discovery can create a minimal identity, but cannot overwrite verified metadata.
|
|
const identity = { provider, uid: instrument.uid };
|
|
const payload = JSON.parse(
|
|
JSON.stringify({
|
|
uid: instrument.uid,
|
|
figi: instrument.figi,
|
|
ticker: instrument.ticker,
|
|
name: instrument.name,
|
|
isin: instrument.isin,
|
|
classCode: instrument.classCode,
|
|
instrumentKind: instrument.instrumentKind,
|
|
apiTradeAvailableFlag: instrument.apiTradeAvailableFlag,
|
|
currency: instrument.currency,
|
|
brand: instrument.brand
|
|
? { logoName: instrument.brand.logoName }
|
|
: undefined,
|
|
}),
|
|
) as Prisma.InputJsonObject;
|
|
await tx.tbankInstrument.upsert({
|
|
where: { provider_uid: identity },
|
|
update: {},
|
|
create: {
|
|
...identity,
|
|
metadata: payload,
|
|
instrumentType: input.instrumentType,
|
|
figi: instrument.figi,
|
|
ticker: instrument.ticker,
|
|
fetchedAt: now,
|
|
},
|
|
});
|
|
const key = {
|
|
provider,
|
|
instrumentUid: instrument.uid,
|
|
source: provider,
|
|
kind: 'catalog',
|
|
rangeKey: input.instrumentType,
|
|
};
|
|
const previous = await tx.tbankInstrumentSource.findUnique({
|
|
where: { provider_instrumentUid_source_kind_rangeKey: key },
|
|
});
|
|
await tx.tbankInstrumentSource.upsert({
|
|
where: { provider_instrumentUid_source_kind_rangeKey: key },
|
|
create: { ...key, payload, fetchedAt: now },
|
|
update: {
|
|
payload,
|
|
fetchedAt: now,
|
|
dataRevision: {
|
|
increment: isDeepStrictEqual(previous?.payload, payload) ? UNCHANGED_REVISION_INCREMENT : REVISION_INCREMENT,
|
|
},
|
|
},
|
|
});
|
|
}
|
|
const instrumentUids = instruments.map((instrument) => instrument.uid);
|
|
if (input.exactTicker) {
|
|
const key = {
|
|
provider,
|
|
identifierType: `${input.instrumentType}-ticker`,
|
|
identifier: input.query,
|
|
};
|
|
const previous = await tx.tbankInstrumentLookup.findUnique({
|
|
where: { provider_identifierType_identifier: key },
|
|
});
|
|
await tx.tbankInstrumentLookup.upsert({
|
|
where: { provider_identifierType_identifier: key },
|
|
create: { ...key, matchedUids: instrumentUids, fetchedAt: now },
|
|
update: {
|
|
matchedUids: instrumentUids,
|
|
fetchedAt: now,
|
|
dataRevision: {
|
|
increment: isDeepStrictEqual(
|
|
previous?.matchedUids,
|
|
instrumentUids,
|
|
)
|
|
? UNCHANGED_REVISION_INCREMENT
|
|
: REVISION_INCREMENT,
|
|
},
|
|
},
|
|
});
|
|
return true;
|
|
}
|
|
const key = {
|
|
provider,
|
|
instrumentType: input.instrumentType,
|
|
query: input.query,
|
|
};
|
|
const previous = await tx.tbankCatalogDiscovery.findUnique({
|
|
where: { provider_instrumentType_query: key },
|
|
});
|
|
await tx.tbankCatalogDiscovery.upsert({
|
|
where: { provider_instrumentType_query: key },
|
|
create: { ...key, instrumentUids, fetchedAt: now },
|
|
update: {
|
|
instrumentUids,
|
|
fetchedAt: now,
|
|
dataRevision: {
|
|
increment: isDeepStrictEqual(
|
|
previous?.instrumentUids,
|
|
instrumentUids,
|
|
)
|
|
? UNCHANGED_REVISION_INCREMENT
|
|
: REVISION_INCREMENT,
|
|
},
|
|
},
|
|
});
|
|
return true;
|
|
});
|
|
}
|
|
|
|
finish(attempt: TbankSyncAttempt, success: boolean) {
|
|
return this.prisma.$transaction(async (tx) => {
|
|
await tx.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: true },
|
|
},
|
|
data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken },
|
|
});
|
|
await tx.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: false },
|
|
},
|
|
data: {
|
|
state: success ? 'complete' : 'failed',
|
|
errorCode: success ? null : TbankErrorCode.Unavailable,
|
|
},
|
|
});
|
|
});
|
|
}
|
|
}
|
|
|
|
function sourceProvider(input: CatalogSource) {
|
|
return input.kind === 'yield' || input.kind === 'bond-events'
|
|
? 'moex'
|
|
: input.kind === 'rating'
|
|
? 'ncr'
|
|
: 'tbank';
|
|
}
|