diff --git a/src/integrations/tbank/tbank.client.ts b/src/integrations/tbank/tbank.client.ts index eb57f01..24cd687 100644 --- a/src/integrations/tbank/tbank.client.ts +++ b/src/integrations/tbank/tbank.client.ts @@ -388,7 +388,12 @@ export class TbankClient { ).then((response) => response.instrument ?? null); } - getEtf(token: string, id: string, idType = InstrumentIdType.Uid, notFound = false): Promise { + getEtf( + token: string, + id: string, + idType = InstrumentIdType.Uid, + notFound = false, + ): Promise> { return this.call<{ instrument?: TbankShare }>(token, this.sparseInstruments, 'etfBy', { idType, id }, notFound) .then((response) => response.instrument ?? null); } diff --git a/src/modules/investments/portfolio/application/portfolio.service.ts b/src/modules/investments/portfolio/application/portfolio.service.ts index 048627c..b001098 100644 --- a/src/modules/investments/portfolio/application/portfolio.service.ts +++ b/src/modules/investments/portfolio/application/portfolio.service.ts @@ -238,7 +238,7 @@ export class PortfolioService { }; } - request(userId: string, revision: string, accountId: string, force = false, walkStartedAt = new Date()) { + request(userId: string, revision: string, accountId: string, force = false, walkStartedAt?: Date) { return this.queueSnapshot(userId, revision, accountId, force, walkStartedAt); } diff --git a/src/modules/investments/portfolio/application/position-classification.service.spec.ts b/src/modules/investments/portfolio/application/position-classification.service.spec.ts index 75631e0..4b26d44 100644 --- a/src/modules/investments/portfolio/application/position-classification.service.spec.ts +++ b/src/modules/investments/portfolio/application/position-classification.service.spec.ts @@ -6,7 +6,7 @@ import { InstrumentIdType } from '../../../../integrations/tbank/tbank-protocol' import { PositionClassificationService } from './position-classification.service'; describe('PositionClassificationService', () => { - const position = { instrumentType: 'share', instrumentUid: 'uid', figi: 'figi' }; + const position = { instrumentType: 'share', instrumentUid: 'uid', figi: 'figi', quantity: { units: 1 } }; const saved = { sector: 'Old', sectorFetchedAt: new Date('2026-10-30T12:00:00Z'), assetUid: 'asset', brandUid: 'brand', brandName: 'Old brand', brandFetchedAt: new Date('2026-10-30T12:00:00Z') }; const tbank = { getShare: jest.fn(), getAssetBrand: jest.fn(), getBond: jest.fn(), getEtf: jest.fn() }; const repository = { read: jest.fn(), save: jest.fn() }; @@ -26,6 +26,10 @@ describe('PositionClassificationService', () => { tbank.getShare.mockResolvedValue({ uid: 'uid', figi: 'figi', sector: ' Tech ', assetUid: 'asset' }); const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); expect(result.values().next().value).toEqual({ sector: 'Tech', issuer: { id: 'new-brand', name: 'New brand' } }); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), { + sector: 'Tech', sectorFetchedAt: new Date('2026-10-31T12:00:00Z'), + brandUid: 'new-brand', brandName: 'New brand', brandFetchedAt: new Date('2026-10-31T12:00:00Z'), + }); }); it('accepts a matching FIGI after UID NOT_FOUND', async () => { @@ -35,14 +39,15 @@ describe('PositionClassificationService', () => { expect(result.values().next().value?.sector).toBe('Energy'); }); - it('preserves saved values on mismatched instrument and database failure', async () => { + it('preserves saved values on mismatched instrument and returns refreshed values on database failure', async () => { tbank.getShare.mockResolvedValue({ uid: 'wrong', sector: 'Energy' }); const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); expect(result.values().next().value?.sector).toBe('Old'); repository.save.mockRejectedValue(new Error('db unavailable')); tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'asset' }); const fallback = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); - expect(fallback.values().next().value?.sector).toBe('Old'); + expect(fallback.values().next().value?.sector).toBe('Energy'); + expect(logger.event).toHaveBeenCalled(); }); it('does not retain an issuer from a previous asset when the asset changes', async () => { @@ -54,10 +59,74 @@ describe('PositionClassificationService', () => { expect(result.values().next().value?.issuer).toEqual({ id: 'other-brand', name: 'Other brand' }); }); + it('does not restart the brand clock when the brand fetch fails after an asset change', async () => { + repository.read.mockResolvedValue({ ...saved, brandFetchedAt: new Date('2026-10-31T12:00:00Z') }); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'other' }); + tbank.getAssetBrand.mockRejectedValue(new Error('upstream unavailable')); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value?.issuer).toBeNull(); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + assetUid: 'other', brandUid: null, brandName: null, brandFetchedAt: null, + })); + }); + + it('does not return the previous asset brand when saving a changed asset fails', async () => { + repository.save.mockRejectedValue(new Error('db unavailable')); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'other' }); + tbank.getAssetBrand.mockRejectedValue(new Error('upstream unavailable')); + const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value).toEqual({ sector: 'Energy', issuer: null }); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + assetUid: 'other', brandUid: null, brandName: null, brandFetchedAt: null, + })); + expect(logger.event).toHaveBeenCalled(); + }); + + it('records a fetched brand as null when a fresh instrument has no asset_uid', async () => { + repository.read.mockResolvedValue(null); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy' }); + await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + brandUid: null, brandName: null, brandFetchedAt: new Date('2026-10-31T12:00:00Z'), + })); + }); + + it('treats a NOT_FOUND answer as fetched and does not refetch it an hour later', async () => { + tbank.getShare.mockRejectedValue(new NotFoundException()); + await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + sector: null, sectorFetchedAt: new Date('2026-10-31T12:00:00Z'), + })); + repository.read.mockResolvedValue({ ...saved, sector: null, sectorFetchedAt: new Date('2026-10-31T12:00:00Z'), assetUid: null, brandUid: null, brandName: null, brandFetchedAt: new Date('2026-10-31T12:00:00Z') }); + tbank.getShare.mockClear(); + await service.refresh('token', [position], new Date('2026-10-31T13:00:00Z')); + expect(tbank.getShare).not.toHaveBeenCalled(); + }); + + it('records a missing brand and restarts its clock when NOT_FOUND has no saved asset UID', async () => { + repository.read.mockResolvedValue(null); + tbank.getShare.mockRejectedValue(new NotFoundException()); + const startedAt = new Date('2026-10-31T12:00:00Z'); + const result = await service.refresh('token', [position], startedAt); + expect(result.values().next().value).toEqual({ sector: null, issuer: null }); + expect(repository.save).toHaveBeenCalledWith(expect.any(String), expect.objectContaining({ + brandUid: null, brandName: null, brandFetchedAt: startedAt, + })); + expect(tbank.getAssetBrand).not.toHaveBeenCalled(); + }); + + it('does not leak one refresh call\'s invalid token into another call', async () => { + tbank.getShare.mockRejectedValueOnce(new Error(TbankErrorCode.InvalidToken)) + .mockResolvedValueOnce({ uid: 'uid', figi: 'figi', sector: 'Energy', assetUid: 'asset' }); + await expect(service.refresh('first', [position], new Date('2026-10-31T12:00:00Z'))).rejects.toThrow(); + const result = await service.refresh('second', [position], new Date('2026-10-31T12:00:00Z')); + expect(result.values().next().value?.sector).toBe('Energy'); + }); + it('propagates an invalid token before publishing a classification', async () => { tbank.getShare.mockRejectedValue(new Error(TbankErrorCode.InvalidToken)); - await expect(service.refresh('token', [position], new Date('2026-10-31T12:00:00Z'))) - .rejects.toThrow(TbankErrorCode.InvalidToken); + const result = service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); + await expect(result).rejects.toThrow(TbankErrorCode.InvalidToken); expect(repository.save).not.toHaveBeenCalled(); }); @@ -68,25 +137,36 @@ describe('PositionClassificationService', () => { expect(tbank.getAssetBrand).not.toHaveBeenCalled(); }); - it('clears the former issuer if the new asset lookup fails', async () => { - repository.read.mockResolvedValue({ ...saved, brandFetchedAt: new Date('2026-10-31T12:00:00Z') }); - tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'other' }); - tbank.getAssetBrand.mockRejectedValue(new Error('upstream unavailable')); - const result = await service.refresh('token', [position], new Date('2026-10-31T12:00:00Z')); - expect(result.values().next().value?.issuer).toBeNull(); - }); - - it('skips unsupported types and shares a brand lookup across instruments', async () => { + it('skips unsupported types, zero-quantity positions, and shares a brand lookup across instruments', async () => { repository.read.mockResolvedValue(null); tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'asset' }); tbank.getBond.mockResolvedValue({ uid: 'bond', sector: 'Energy', assetUid: 'asset' }); const result = await service.refresh('token', [ - { instrumentType: 'currency', instrumentUid: 'cash', figi: 'cash' }, + { instrumentType: 'currency', instrumentUid: 'cash', figi: 'cash', quantity: { units: 1 } }, + { instrumentType: 'share', instrumentUid: 'empty', figi: 'empty', quantity: {} }, + { instrumentType: 'Share', instrumentUid: 'upper', figi: 'upper', quantity: { units: 1 } }, position, - { instrumentType: 'bond', instrumentUid: 'bond', figi: 'bond-figi' }, + { instrumentType: 'bond', instrumentUid: 'bond', figi: 'bond-figi', quantity: { units: 1 } }, ], new Date('2026-10-31T12:00:00Z')); - expect(result.size).toBe(2); + expect(result.size).toBe(3); expect(tbank.getAssetBrand).toHaveBeenCalledTimes(1); expect(tbank.getEtf).not.toHaveBeenCalled(); }); + + it('skips duplicate instrument keys and retries a failed asset lookup for another instrument in the same walk', async () => { + repository.read.mockResolvedValue(null); + tbank.getShare.mockResolvedValue({ uid: 'uid', sector: 'Energy', assetUid: 'asset' }); + tbank.getBond.mockResolvedValue({ uid: 'bond', sector: 'Energy', assetUid: 'asset' }); + tbank.getAssetBrand.mockRejectedValueOnce(new Error('upstream unavailable')) + .mockResolvedValueOnce({ assetUid: 'asset', brand: { uid: 'new-brand', name: 'New brand' } }); + const result = await service.refresh('token', [position, position, + { instrumentType: 'bond', instrumentUid: 'bond', figi: 'bond-figi', quantity: { units: 1 } }, + ], new Date('2026-10-31T12:00:00Z')); + expect(result.size).toBe(2); + expect(tbank.getShare).toHaveBeenCalledTimes(1); + expect(tbank.getAssetBrand).toHaveBeenCalledTimes(2); + expect([...result.values()].map(({ issuer }) => issuer)).toEqual([ + null, { id: 'new-brand', name: 'New brand' }, + ]); + }); }); diff --git a/src/modules/investments/portfolio/application/position-classification.service.ts b/src/modules/investments/portfolio/application/position-classification.service.ts index e2a0ff2..1832f6a 100644 --- a/src/modules/investments/portfolio/application/position-classification.service.ts +++ b/src/modules/investments/portfolio/application/position-classification.service.ts @@ -1,7 +1,9 @@ import { Injectable, NotFoundException } from '@nestjs/common'; +import { isZero } from '../../../../common/decimal'; import { StructuredLogger } from '../../../../common/logging/structured-logger.service'; import { MILLISECONDS_PER_DAY } from '../../../../common/time'; +import { decimal } from '../../../../integrations/tbank/decimal'; import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; import { InstrumentIdType } from '../../../../integrations/tbank/tbank-protocol'; import { metadataKey, TbankClient, type TbankPortfolio } from '../../../../integrations/tbank/tbank.api'; @@ -11,15 +13,10 @@ export type Classification = { issuer: null | { id: string; name: string }; sect type Position = NonNullable[number]; type Instrument = { assetUid?: string; figi?: string; sector?: string; uid?: string }; type Row = NonNullable>>; - -const dayMs = MILLISECONDS_PER_DAY; type Update = Parameters[1]; @Injectable() export class PositionClassificationService { - private readonly instruments = new Map>(); - private readonly assets = new Map>(); - constructor( private readonly tbank: TbankClient, private readonly repository: PositionClassificationRepository, @@ -28,106 +25,112 @@ export class PositionClassificationService { async refresh(token: string, positions: Position[], walkStartedAt: Date): Promise> { const result = new Map(); - const assets = new Map>(); + const inFlight = new Map>(); for (const position of positions) { - if (!['bond', 'etf', 'share'].includes(position.instrumentType ?? '')) continue; + if (!['bond', 'etf', 'share'].includes((position.instrumentType ?? '').toLowerCase())) continue; + if (isZero(decimal(position.quantity ?? {}))) continue; const key = metadataKey(position); if (result.has(key)) continue; - result.set(key, await this.refreshOne(token, position, key, walkStartedAt, assets)); + result.set(key, await this.refreshOne(token, position, key, walkStartedAt, inFlight)); } return result; } private async refreshOne(token: string, position: Position, key: string, startedAt: Date, - assets: Map>): Promise { + inFlight: Map>): Promise { let saved: null | Row = null; + const update: Update = {}; try { saved = await this.repository.read(key); - const update: Update = {}; - const due = (at: Date | null | undefined, value: null | string | undefined) => - !value || !at || startedAt.valueOf() - at.valueOf() >= dayMs; - const sectorDue = due(saved?.sectorFetchedAt, saved?.sector); - const brandDue = due(saved?.brandFetchedAt, saved?.brandUid && saved.brandName); - const assetUid = sectorDue || (brandDue && !saved?.assetUid) - ? await this.refreshInstrument(token, position, key, sectorDue, saved?.assetUid ?? null, update) - : saved?.assetUid ?? null; + const due = (at: Date | null | undefined) => !at || startedAt.valueOf() - at.valueOf() >= MILLISECONDS_PER_DAY; + const sectorDue = due(saved?.sectorFetchedAt); + const brandDue = due(saved?.brandFetchedAt); + let assetUid = saved?.assetUid ?? null; + let instrumentFresh = false; + if (sectorDue || (brandDue && !saved?.assetUid)) { + const instrument = await this.refreshInstrument(token, position, sectorDue, assetUid, update, startedAt); + assetUid = instrument.assetUid; + instrumentFresh = instrument.fresh; + } const assetChanged = 'assetUid' in update && update.assetUid !== saved?.assetUid; if (assetChanged) { update.brandUid = null; update.brandName = null; - update.brandFetchedAt = new Date(); + update.brandFetchedAt = null; } if (brandDue || assetChanged) { - await this.refreshBrand(token, assetUid, update, assets); + await this.refreshBrand(token, assetUid, update, startedAt, inFlight, instrumentFresh); } if (Object.keys(update).length) saved = await this.repository.save(key, update); } catch (error) { if (this.invalidToken(error)) throw error; - this.failed(); + this.failed(error); } + const current = { ...saved, ...update }; return { - issuer: saved?.brandUid && saved.brandName ? { id: saved.brandUid, name: saved.brandName } : null, - sector: saved?.sector ?? null, + issuer: current.brandUid && current.brandName ? { id: current.brandUid, name: current.brandName } : null, + sector: current.sector ?? null, }; } - private async refreshInstrument(token: string, position: Position, key: string, sectorDue: boolean, savedUid: null | string, update: Update) { + private async refreshInstrument(token: string, position: Position, sectorDue: boolean, savedUid: null | string, + update: Update, startedAt: Date): Promise<{ assetUid: null | string; fresh: boolean }> { try { - const lookup = await this.instrument(token, position, key); - if (lookup && !this.matches(lookup)) return savedUid; - if (lookup) return this.acceptInstrument(lookup.found, savedUid, sectorDue, update); - if (sectorDue) this.clearSector(update); + const lookup = await this.instrument(token, position); + if (lookup && !this.matches(lookup)) return { assetUid: savedUid, fresh: false }; + if (lookup) return { assetUid: this.acceptInstrument(lookup.found, savedUid, sectorDue, update, startedAt), fresh: true }; + if (sectorDue) this.clearSector(update, startedAt); } catch (error) { if (this.invalidToken(error)) throw error; if (error instanceof NotFoundException) { - if (sectorDue) this.clearSector(update); - } else this.failed(); + if (sectorDue) this.clearSector(update, startedAt); + } else this.failed(error); } - return savedUid; + return { assetUid: savedUid, fresh: false }; } - private clearSector(update: Update) { + private clearSector(update: Update, startedAt: Date) { update.sector = null; - update.sectorFetchedAt = new Date(); + update.sectorFetchedAt = startedAt; } - private acceptInstrument(found: Instrument, savedUid: null | string, sectorDue: boolean, update: Update) { + private acceptInstrument(found: Instrument, savedUid: null | string, sectorDue: boolean, update: Update, startedAt: Date) { const assetUid = found.assetUid?.trim() || null; if (assetUid !== savedUid) update.assetUid = assetUid; - if (sectorDue) this.setSector(update, found.sector); + if (sectorDue) this.setSector(update, found.sector, startedAt); return assetUid; } - private setSector(update: Update, sector: string | undefined) { + private setSector(update: Update, sector: string | undefined, startedAt: Date) { update.sector = sector?.trim() || null; - update.sectorFetchedAt = new Date(); + update.sectorFetchedAt = startedAt; } - private async refreshBrand(token: string, assetUid: null | string, update: Update, - assets: Map>) { + private async refreshBrand(token: string, assetUid: null | string, update: Update, startedAt: Date, + inFlight: Map>, instrumentFresh: boolean) { if (!assetUid) { - if ('assetUid' in update || 'sectorFetchedAt' in update) this.clearBrand(update); + if (instrumentFresh || 'assetUid' in update || 'sectorFetchedAt' in update) this.clearBrand(update, startedAt); return; } try { - const asset = await this.cachedAsset(token, assetUid, assets); + const asset = await this.dedupe(inFlight, assetUid, () => this.tbank.getAssetBrand(token, assetUid)); if (asset && asset.assetUid !== assetUid) return; const uid = asset?.brand?.uid?.trim() || null; const name = asset?.brand?.name?.trim() || null; update.brandUid = uid && name ? uid : null; update.brandName = uid && name ? name : null; - update.brandFetchedAt = new Date(); + update.brandFetchedAt = startedAt; } catch (error) { if (this.invalidToken(error)) throw error; - if (error instanceof NotFoundException) this.clearBrand(update); - else this.failed(); + if (error instanceof NotFoundException) this.clearBrand(update, startedAt); + else this.failed(error); } } - private clearBrand(update: Update) { + private clearBrand(update: Update, startedAt: Date) { update.brandUid = null; update.brandName = null; - update.brandFetchedAt = new Date(); + update.brandFetchedAt = startedAt; } private matches(lookup: { found: Instrument; id: string; idType: InstrumentIdType }) { @@ -138,55 +141,44 @@ export class PositionClassificationService { return error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken); } - private async instrument(token: string, position: Position, key: string) { - return this.inFlight(this.instruments, key, async () => { - const lookup = (id: string, type: InstrumentIdType) => { - switch (position.instrumentType) { - case 'bond': return this.tbank.getBond(token, id, type, true); - case 'share': return this.tbank.getShare(token, id, type, true); - default: return this.tbank.getEtf(token, id, type, true); - } - }; - const uid = position.instrumentUid?.trim(); - const figi = position.figi?.trim(); - if (uid) { - try { - const found = await lookup(uid, InstrumentIdType.Uid); - if (found) return { found, id: uid, idType: InstrumentIdType.Uid }; - } catch (error) { - if (!(error instanceof NotFoundException)) throw error; - } + private async instrument(token: string, position: Position) { + const type = (position.instrumentType ?? '').toLowerCase(); + const lookup = (id: string, idType: InstrumentIdType) => { + switch (type) { + case 'bond': return this.tbank.getBond(token, id, idType, true); + case 'share': return this.tbank.getShare(token, id, idType, true); + default: return this.tbank.getEtf(token, id, idType, true); } - if (figi) { - const found = await lookup(figi, InstrumentIdType.Figi); - return found ? { found, id: figi, idType: InstrumentIdType.Figi } : null; + }; + const uid = position.instrumentUid?.trim(); + const figi = position.figi?.trim(); + if (uid) { + try { + const found = await lookup(uid, InstrumentIdType.Uid); + if (found) return { found, id: uid, idType: InstrumentIdType.Uid }; + } catch (error) { + if (!(error instanceof NotFoundException)) throw error; } - return null; - }); - } - - private asset(token: string, uid: string) { - return this.inFlight(this.assets, uid, () => this.tbank.getAssetBrand(token, uid)); - } - - private cachedAsset(token: string, uid: string, assets: Map>) { - let pending = assets.get(uid); - if (!pending) { - pending = this.asset(token, uid); - assets.set(uid, pending); } - return pending; + if (figi) { + const found = await lookup(figi, InstrumentIdType.Figi); + return found ? { found, id: figi, idType: InstrumentIdType.Figi } : null; + } + return null; } - private inFlight(map: Map>, key: string, run: () => Promise): Promise { + private dedupe(map: Map>, key: string, run: () => Promise): Promise { const pending = map.get(key); - if (pending) return pending; - const promise = run().finally(() => map.delete(key)); + if (pending) return pending as Promise; + const promise = run().catch((error: unknown) => { + map.delete(key); + throw error; + }); map.set(key, promise); return promise; } - private failed() { - this.logger.event('warn', 'portfolio.classification.failed', { code: TbankErrorCode.Unavailable }); + private failed(error: unknown) { + this.logger.event('warn', 'portfolio.classification.failed', { code: error instanceof Error ? error.name : 'Unknown' }); } } diff --git a/src/modules/investments/portfolio/infrastructure/position-classification.repository.ts b/src/modules/investments/portfolio/infrastructure/position-classification.repository.ts index 5efd2ad..937cc18 100644 --- a/src/modules/investments/portfolio/infrastructure/position-classification.repository.ts +++ b/src/modules/investments/portfolio/infrastructure/position-classification.repository.ts @@ -14,7 +14,7 @@ export class PositionClassificationRepository { save(key: string, data: { assetUid?: null | string; - brandFetchedAt?: Date; + brandFetchedAt?: Date | null; brandName?: null | string; brandUid?: null | string; sector?: null | string;