332 lines
10 KiB
TypeScript
332 lines
10 KiB
TypeScript
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 { TbankSyncAttempt } from './tbank-access.service';
|
|
import type { TbankInstrument } from './tbank.client';
|
|
|
|
import { PrismaService } from '../../prisma/prisma.service';
|
|
import { TbankErrorCode } from './tbank-error-code';
|
|
|
|
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<string, ReturnType<typeof selectMetadata>>();
|
|
}
|
|
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<string>,
|
|
) {
|
|
return this.prisma.$transaction(async (tx) => {
|
|
const valid = await tx.tbankConnection.updateMany({
|
|
where: { ...attemptOwner, invalidToken: false },
|
|
data: { membershipRevision: { increment: 0 } },
|
|
});
|
|
if (!valid.count) return null;
|
|
const scope = {
|
|
...attemptOwner,
|
|
scopeKey: `metadata:${metadataKey(identity)}`,
|
|
};
|
|
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: '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 tx.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: false },
|
|
},
|
|
data: { state: 'complete', progress: {} },
|
|
});
|
|
if (!claimed.count) 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 this.prisma.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
...(success ? { connection: { invalidToken: false } } : {}),
|
|
},
|
|
data: {
|
|
state: success ? 'complete' : 'failed',
|
|
errorCode: success ? null : TbankErrorCode.Unavailable,
|
|
},
|
|
});
|
|
if (success) {
|
|
await this.prisma.tbankSyncWork.updateMany({
|
|
where: {
|
|
...attempt,
|
|
state: 'running',
|
|
connection: { invalidToken: true },
|
|
},
|
|
data: { state: 'failed', errorCode: TbankErrorCode.InvalidToken },
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
function selectMetadata(
|
|
identity: MetadataIdentity,
|
|
lookups: ReadonlyMap<string, MetadataLookupRow>,
|
|
instruments: ReadonlyMap<string, MetadataRow>,
|
|
) {
|
|
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('/'),
|
|
};
|
|
}
|