import { Injectable } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import type { TbankInstrumentLookup as MetadataLookupRow, TbankInstrument as MetadataRow, Prisma, } from '../../../generated/prisma/client'; import type { TbankInstrument } from './tbank.client'; import { PrismaService } from '../../prisma/prisma.service'; import { isJoinableRunningAttempt, type TbankSyncAttempt } from './tbank-access.service'; import { TbankErrorCode } from './tbank-error-code'; import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim'; const FIRST_BATCH_OFFSET = 0; const UNCHANGED_REVISION_INCREMENT = 0; const REVISION_INCREMENT = 1; const MISSING_DATA_REVISION = 0; export type MetadataIdentity = { figi?: null | string; instrumentUid?: null | string; }; export type MetadataLookup = { data: null | TbankInstrument; id: string; type: 'figi' | 'uid'; }; const provider = 'tbank'; // Keep IN lists below the SQLite/Prisma parameter limit, including fixed predicates. const metadataReadBatchSize = 400; export function metadataKey(identity: MetadataIdentity) { return JSON.stringify([ identity.instrumentUid?.trim() || null, identity.figi?.trim() || null, ]); } @Injectable() export class InstrumentMetadataRepository { constructor(private readonly prisma: PrismaService) {} async readMany(identities: MetadataIdentity[]) { const identifiers = identities.flatMap((identity) => [ ...(identity.instrumentUid?.trim() ? [{ identifierType: 'uid', identifier: identity.instrumentUid.trim() }] : []), ...(identity.figi?.trim() ? [{ identifierType: 'figi', identifier: identity.figi.trim() }] : []), ]); if (!identifiers.length) { return new Map>(); } return this.prisma.$transaction(async (tx) => { const lookups: MetadataLookupRow[] = []; for (const identifierType of ['uid', 'figi']) { const values = [ ...new Set( identifiers .filter((entry) => entry.identifierType === identifierType) .map((entry) => entry.identifier), ), ]; for ( let offset = FIRST_BATCH_OFFSET; offset < values.length; offset += metadataReadBatchSize ) { lookups.push( ...(await tx.tbankInstrumentLookup.findMany({ where: { provider, identifierType, identifier: { in: values.slice(offset, offset + metadataReadBatchSize), }, }, })), ); } } const uids = [ ...new Set(lookups.flatMap((row) => row.matchedUids as string[])), ]; const instruments: MetadataRow[] = []; for ( let offset = FIRST_BATCH_OFFSET; offset < uids.length; offset += metadataReadBatchSize ) { instruments.push( ...(await tx.tbankInstrument.findMany({ where: { provider, uid: { in: uids.slice(offset, offset + metadataReadBatchSize) }, }, })), ); } const byIdentifier = new Map( lookups.map((row) => [`${row.identifierType}:${row.identifier}`, row]), ); const byUid = new Map(instruments.map((row) => [row.uid, row])); return new Map( identities.map((identity) => [ metadataKey(identity), selectMetadata(identity, byIdentifier, byUid), ]), ); }); } async reserve( attemptOwner: { revision: string; userId: string }, identity: MetadataIdentity, runId: string, active: ReadonlySet, ) { return this.prisma.$transaction(async (tx) => { const valid = await lockConnectionGeneration(tx, attemptOwner); if (!valid) return null; const scope = { ...attemptOwner, scopeKey: `metadata:${metadataKey(identity)}`, }; 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: 'metadata', ...identity }, 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 }; }); } work(userId: string, revision: string, identity: MetadataIdentity) { return this.prisma.tbankSyncWork.findUnique({ where: { userId_revision_scopeKey: { userId, revision, scopeKey: `metadata:${metadataKey(identity)}`, }, }, select: { id: true, state: true, errorCode: true }, }); } publish(attempt: TbankSyncAttempt, results: MetadataLookup[], now: Date) { return this.prisma.$transaction(async (tx) => { const claimed = await claimSyncWork(tx, attempt, { state: 'complete', progress: {}, }); if (!claimed) return false; const aliases = new Map( results.map((result) => [`${result.type}:${result.id}`, result]), ); for (const result of results) { if (!result.data) continue; const uid = result.data.uid?.trim() || (result.type === 'uid' ? result.id : null); const figi = result.data.figi?.trim() || (result.type === 'figi' ? result.id : null); for (const [type, id] of [ ['uid', uid], ['figi', figi], ] as const) { if (id && uid && !aliases.has(`${type}:${id}`)) { aliases.set(`${type}:${id}`, { type, id, data: { ...result.data, uid }, }); } } } for (const result of aliases.values()) { const instrument = result.data; const uid = instrument?.uid?.trim() || (result.type === 'uid' ? result.id : null); // FIGI is an alias, never a fabricated provider UID. if (instrument && !uid) { throw new Error('Metadata response has no instrument UID'); } if (instrument && uid) { const key = { provider, uid }; const previous = await tx.tbankInstrument.findUnique({ where: { provider_uid: key }, }); const payload = JSON.parse( JSON.stringify(instrument), ) as Prisma.InputJsonObject; const data = { metadata: payload, figi: instrument.figi ?? (result.type === 'figi' ? result.id : null), ticker: instrument.ticker ?? null, instrumentType: instrument.instrumentType ?? '', fetchedAt: now, }; await tx.tbankInstrument.upsert({ where: { provider_uid: key }, create: { ...key, ...data }, update: { ...data, dataRevision: { increment: isDeepStrictEqual(previous?.metadata, payload) ? UNCHANGED_REVISION_INCREMENT : REVISION_INCREMENT, }, }, }); } const key = { provider, identifierType: result.type, identifier: result.id, }; const matchedUids = instrument && uid ? [uid] : []; const previous = await tx.tbankInstrumentLookup.findUnique({ where: { provider_identifierType_identifier: key }, }); await tx.tbankInstrumentLookup.upsert({ where: { provider_identifierType_identifier: key }, create: { ...key, matchedUids, fetchedAt: now }, update: { matchedUids, fetchedAt: now, dataRevision: { increment: isDeepStrictEqual(previous?.matchedUids, matchedUids) ? UNCHANGED_REVISION_INCREMENT : REVISION_INCREMENT, }, }, }); } return true; }); } async finishObserver(attempt: TbankSyncAttempt, success: boolean) { await claimSyncWork( this.prisma, attempt, { state: success ? 'complete' : 'failed', errorCode: success ? null : TbankErrorCode.Unavailable, }, { withoutConnectionFence: !success }, ); if (success) { // Fails sibling attempts whose connection was invalidated mid-flight. await this.prisma.tbankSyncWork.updateMany({ where: { ...attempt, state: 'running', connection: { invalidToken: true }, }, data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken }, }); } } } function selectMetadata( identity: MetadataIdentity, lookups: ReadonlyMap, instruments: ReadonlyMap, ) { const resolve = (type: string, id: null | string | undefined) => { const lookup = lookups.get(`${type}:${id?.trim()}`); const uid = (lookup?.matchedUids as string[] | undefined)?.[0]; const instrument = uid ? instruments.get(uid) : undefined; const data = instrument ? (instrument.metadata as TbankInstrument) : null; return { lookup, instrument, data }; }; const instrumentUid = identity.instrumentUid?.trim(); const instrumentFigi = identity.figi?.trim(); const uid = resolve('uid', instrumentUid); const figi = resolve('figi', instrumentFigi); const usefulUid = Boolean(uid.data?.name?.trim() || uid.data?.ticker?.trim()); const used = usefulUid || !instrumentFigi ? [uid] : instrumentUid ? [uid, figi] : [figi]; const complete = used.every((row) => Boolean(row.lookup)); const fetchedAt = used .flatMap((row) => (row.lookup ? [row.lookup.fetchedAt] : [])) .sort((a, b) => a.valueOf() - b.valueOf())[0] ?? null; return { data: usefulUid ? uid.data : (figi.data ?? uid.data), complete, fresh: complete, fetchedAt, dataRevision: used .map( (row) => `${row.lookup?.dataRevision ?? MISSING_DATA_REVISION}:${row.instrument?.dataRevision ?? MISSING_DATA_REVISION}`, ) .join('/'), }; }