import { Injectable } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import type { Prisma } from '../../../../../generated/prisma/client'; import type { TbankBond, TbankShare, TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api'; import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; 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[]) { return this.prisma.tbankInstrumentSource.findMany({ where: { provider, OR: inputs.map((input) => ({ instrumentUid: input.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, ) { 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 ( previous?.state === 'running' && previous.processRunId === runId && active.has(previous.attemptId) ) { 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'; }