From aee330c6c2ef82f05cda4eb0923d332a9b787257 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Fri, 9 Oct 2026 19:56:14 +0300 Subject: [PATCH] feat: reconcile hourly investment resource states --- .../brokerage-accounts.module.ts | 4 +- .../brokerage-accounts.service.spec.ts | 37 +++++- .../brokerage-accounts.service.ts | 15 ++- .../income/income-analytics.service.spec.ts | 2 +- .../instruments/application/candle.service.ts | 5 +- .../application/catalog-discovery.service.ts | 4 +- .../application/catalog-enrichment.service.ts | 6 +- .../bonds/application/bond-income.service.ts | 4 +- .../application/stock-income.service.ts | 4 +- .../investments/investment-resource.dto.ts | 51 +++++++-- .../investment-resource.mapper.spec.ts | 108 +++++++++++++++++- .../investments/investment-resource.mapper.ts | 107 ++++++++++++----- .../brokerage-operations.service.ts | 9 +- .../operations/operations.module.ts | 3 +- .../portfolio/application/forecast.service.ts | 35 +++++- .../application/portfolio.service.spec.ts | 3 +- .../application/portfolio.service.ts | 14 ++- .../infrastructure/forecast.repository.ts | 35 +++++- .../investments/portfolio/portfolio.module.ts | 3 +- .../tbank-sync-repository.module.ts | 9 ++ .../investments/tbank-sync/tbank-sync.dto.ts | 5 +- .../tbank-sync/tbank-sync.module.ts | 7 +- .../tbank-sync/tbank-sync.repository.ts | 35 ++++++ .../upcoming-events/upcoming-events.module.ts | 3 +- .../upcoming-events.service.ts | 23 +++- 25 files changed, 445 insertions(+), 86 deletions(-) create mode 100644 src/modules/investments/tbank-sync/tbank-sync-repository.module.ts diff --git a/src/modules/investments/brokerage-accounts/brokerage-accounts.module.ts b/src/modules/investments/brokerage-accounts/brokerage-accounts.module.ts index 98a4315..6d1c436 100644 --- a/src/modules/investments/brokerage-accounts/brokerage-accounts.module.ts +++ b/src/modules/investments/brokerage-accounts/brokerage-accounts.module.ts @@ -3,12 +3,12 @@ import { Module } from '@nestjs/common'; import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { AuthModule } from '../../auth/auth.module'; import { PortfolioModule } from '../portfolio/portfolio.module'; -import { TbankSyncModule } from '../tbank-sync/tbank-sync.module'; +import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module'; import { BrokerageAccountsController } from './brokerage-accounts.controller'; import { BrokerageAccountsService } from './brokerage-accounts.service'; @Module({ - imports: [AuthModule, TbankModule, PortfolioModule, TbankSyncModule], + imports: [AuthModule, TbankModule, PortfolioModule, TbankSyncRepositoryModule], controllers: [BrokerageAccountsController], providers: [BrokerageAccountsService], exports: [BrokerageAccountsService], diff --git a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts index 5e38b9d..bfb01ba 100644 --- a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts +++ b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.spec.ts @@ -24,7 +24,7 @@ describe('saved brokerage account projection', () => { it('distinguishes unknown membership from a confirmed empty list', async () => { portfolio.saved.mockResolvedValueOnce({ - membership: { ...meta, availability: 'missing' }, + membership: { ...meta, completeness: 'incomplete' }, accounts: [], }); expect((await service.list('owner')).data).toBeNull(); @@ -66,6 +66,39 @@ describe('saved brokerage account projection', () => { bondPositionCount: 1, }); expect(response.data?.[1].portfolioSummary).toBeNull(); - expect(response.meta.availability).toBe('partial'); + expect(response.meta.availability).toBe('available'); + expect(response.meta.completeness).toBe('complete'); + }); + + it('observes the latest hourly walk without starting another one', async () => { + portfolio.saved.mockResolvedValue({ membership: meta, accounts: [] }); + sync.hourly.mockResolvedValue({ + startedAt: new Date('2026-09-25T12:00:00Z'), + finishedAt: new Date('2026-09-25T12:01:00Z'), + state: 'partial', + sections: [ + { + section: 'portfolio', + accountId: 'a', + errorCode: 'TBANK_UNAVAILABLE', + }, + ], + }); + + const response = await service.list('owner'); + + expect(response.sync).toEqual({ + startedAt: '2026-09-25T12:00:00.000Z', + finishedAt: '2026-09-25T12:01:00.000Z', + state: 'partial', + problems: [ + { + section: 'portfolio', + accountId: 'a', + errorCode: 'TBANK_UNAVAILABLE', + }, + ], + }); + expect(sync.hourly).toHaveBeenCalledWith('owner'); }); }); diff --git a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts index 1c03544..9a003ca 100644 --- a/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts +++ b/src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts @@ -16,7 +16,7 @@ export class BrokerageAccountsService { async list(userId: string): Promise { const saved = await this.portfolio.saved(userId, 'all'); const data: BrokerageAccountDto[] | null = - saved.membership.availability === 'missing' + saved.membership.completeness === 'incomplete' ? null : saved.accounts.map((account) => ({ id: account.id, @@ -55,16 +55,12 @@ export class BrokerageAccountsService { availability: data === null ? 'missing' - : saved.accounts.some((account) => !account.snapshot) - ? 'partial' - : 'complete', + : 'complete', fetchedAt: saved.membership.lastSuccessfulAt ? new Date(saved.membership.lastSuccessfulAt) : null, freshForMs: 3_600_000, - work: saved.membership.sync.id - ? { ...saved.membership.sync, id: saved.membership.sync.id } - : null, + work: null, dependencies: saved.accounts.map((account) => account.meta), }), sync: hourly @@ -72,7 +68,10 @@ export class BrokerageAccountsService { state: hourly.state, startedAt: hourly.startedAt.toISOString(), finishedAt: hourly.finishedAt?.toISOString() ?? null, - sections: hourly.sections, + problems: hourly.sections.map((section) => ({ + ...section, + accountId: section.accountId ?? null, + })), } : null, }; diff --git a/src/modules/investments/income/income-analytics.service.spec.ts b/src/modules/investments/income/income-analytics.service.spec.ts index 03a76dc..c136846 100644 --- a/src/modules/investments/income/income-analytics.service.spec.ts +++ b/src/modules/investments/income/income-analytics.service.spec.ts @@ -48,7 +48,7 @@ describe('IncomeAnalyticsService', () => { expect.objectContaining({ role: 'previous' }), expect.objectContaining({ role: 'year', - availability: 'missing', + availability: 'temporarily-unavailable', unavailableReason: 'CALENDAR_OR_ACCOUNT_HISTORY', }), ]); diff --git a/src/modules/investments/instruments/application/candle.service.ts b/src/modules/investments/instruments/application/candle.service.ts index 5b7194c..812e843 100644 --- a/src/modules/investments/instruments/application/candle.service.ts +++ b/src/modules/investments/instruments/application/candle.service.ts @@ -106,7 +106,7 @@ export class CandleService { true, ['detail'], ); - if (detail.meta.availability === 'complete' && !detail.instrument) { + if (detail.meta.completeness === 'complete' && !detail.instrument) { throw new NotFoundException(); } const now = new Date(); @@ -114,8 +114,7 @@ export class CandleService { ? rangeFor(input, detail.instrument, now) : null; const saved = range ? await this.repository.read(input.uid, range) : null; - const gaps = range && saved ? candleGaps(range, saved.coverage, now) : []; - if (!observe && (!saved || gaps.length)) { + if (!observe && !saved) { await this.request(userId, revision, input, now); } const work = await this.repository.work(userId, revision, scope(input)); diff --git a/src/modules/investments/instruments/application/catalog-discovery.service.ts b/src/modules/investments/instruments/application/catalog-discovery.service.ts index 609e91e..fb3c389 100644 --- a/src/modules/investments/instruments/application/catalog-discovery.service.ts +++ b/src/modules/investments/instruments/application/catalog-discovery.service.ts @@ -121,7 +121,7 @@ export class CatalogDiscoveryService { availability: discovery.data === null ? 'missing' - : sources.every((source) => source.meta.availability === 'complete') + : sources.every((source) => source.meta.completeness === 'complete') ? 'complete' : 'partial', fetchedAt: discovery.meta.lastSuccessfulAt @@ -173,7 +173,7 @@ export class CatalogDiscoveryService { ['detail'], ); const source = sources[0]; - if (source.meta.availability === 'complete' && !source.instrument) { + if (source.meta.completeness === 'complete' && !source.instrument) { throw new NotFoundException(); } const instrument = source.instrument; diff --git a/src/modules/investments/instruments/application/catalog-enrichment.service.ts b/src/modules/investments/instruments/application/catalog-enrichment.service.ts index f97404e..87b5f2f 100644 --- a/src/modules/investments/instruments/application/catalog-enrichment.service.ts +++ b/src/modules/investments/instruments/application/catalog-enrichment.service.ts @@ -288,7 +288,7 @@ export class CatalogEnrichmentService { kinds, ); const first = sources[0]; - if (first.meta.availability === 'complete' && !first.instrument) { + if (first.meta.completeness === 'complete' && !first.instrument) { throw new NotFoundException(); } const values = sources.reduce( @@ -297,7 +297,7 @@ export class CatalogEnrichmentService { ); const dependencies = sources.map((source) => source.meta); const complete = dependencies.every( - (dependency) => dependency.availability === 'complete', + (dependency) => dependency.completeness === 'complete', ); const meta = investmentResourceMeta({ resourceKey: `${instrumentType}/${section}/${uid}`, @@ -322,7 +322,7 @@ export class CatalogEnrichmentService { unavailable: sources .filter( (source) => - source.meta.availability !== 'complete' || + source.meta.completeness !== 'complete' || source.meta.sync.state === 'failed', ) .map((source) => source.input.kind), diff --git a/src/modules/investments/instruments/bonds/application/bond-income.service.ts b/src/modules/investments/instruments/bonds/application/bond-income.service.ts index 5189ed2..675513a 100644 --- a/src/modules/investments/instruments/bonds/application/bond-income.service.ts +++ b/src/modules/investments/instruments/bonds/application/bond-income.service.ts @@ -97,7 +97,7 @@ export class BondIncomeService { ['detail', 'coupons'], { from, to }, ); - if (sources[0].meta.availability === 'complete' && !sources[0].instrument) { + if (sources[0].meta.completeness === 'complete' && !sources[0].instrument) { throw new NotFoundException(); } const events = sources[1].payload?.coupons; @@ -137,7 +137,7 @@ export class BondIncomeService { projected.observedAt, true, ), - meta: { ...result.meta, availability: 'partial' }, + meta: { ...result.meta, completeness: 'incomplete' }, }; } } diff --git a/src/modules/investments/instruments/stocks/application/stock-income.service.ts b/src/modules/investments/instruments/stocks/application/stock-income.service.ts index f98557c..18463be 100644 --- a/src/modules/investments/instruments/stocks/application/stock-income.service.ts +++ b/src/modules/investments/instruments/stocks/application/stock-income.service.ts @@ -66,7 +66,7 @@ export class StockIncomeService { ['detail', 'dividends'], { from, to }, ); - if (sources[0].meta.availability === 'complete' && !sources[0].instrument) { + if (sources[0].meta.completeness === 'complete' && !sources[0].instrument) { throw new NotFoundException(); } const events = sources[1].payload?.dividends; @@ -106,7 +106,7 @@ export class StockIncomeService { projected.observedAt, true, ), - meta: { ...result.meta, availability: 'partial' }, + meta: { ...result.meta, completeness: 'incomplete' }, }; } } diff --git a/src/modules/investments/investment-resource.dto.ts b/src/modules/investments/investment-resource.dto.ts index 5d3eca3..ec666f6 100644 --- a/src/modules/investments/investment-resource.dto.ts +++ b/src/modules/investments/investment-resource.dto.ts @@ -1,9 +1,21 @@ import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; import { IsIn, IsOptional } from 'class-validator'; -export const availabilityValues = ['missing', 'partial', 'complete'] as const; +export const availabilityValues = [ + 'available', + 'initial-loading', + 'temporarily-unavailable', + 'connection-unavailable', +] as const; +export const completenessValues = ['complete', 'incomplete'] as const; export const freshnessValues = ['unknown', 'fresh', 'stale'] as const; -export const syncStateValues = ['idle', 'running', 'failed'] as const; +export const attemptStateValues = [ + 'none', + 'running', + 'succeeded', + 'partial', + 'failed', +] as const; export class InvestmentReadQueryDto { @ApiPropertyOptional({ enum: ['observe'] }) @@ -12,17 +24,31 @@ export class InvestmentReadQueryDto { declare sync?: 'observe'; } -export class InvestmentSyncDto { - @ApiProperty({ type: String, nullable: true }) - declare id: null | string; +export class InvestmentAttemptProblemDto { + @ApiProperty({ enum: ['membership', 'portfolio', 'operations', 'events'] }) + declare section: 'events' | 'membership' | 'operations' | 'portfolio'; - @ApiProperty({ enum: syncStateValues }) - declare state: (typeof syncStateValues)[number]; + @ApiProperty({ type: String, nullable: true }) + declare accountId: null | string; @ApiProperty({ type: String, nullable: true }) declare errorCode: null | string; } +export class InvestmentAttemptDto { + @ApiProperty({ enum: attemptStateValues }) + declare state: (typeof attemptStateValues)[number]; + + @ApiProperty({ type: String, nullable: true, format: 'date-time' }) + declare startedAt: null | string; + + @ApiProperty({ type: String, nullable: true, format: 'date-time' }) + declare finishedAt: null | string; + + @ApiProperty({ type: InvestmentAttemptProblemDto, isArray: true }) + declare problems: InvestmentAttemptProblemDto[]; +} + export class InvestmentRangeDto { @ApiPropertyOptional() accountId?: string; @@ -73,14 +99,21 @@ export class InvestmentDependencyDto { @ApiProperty({ enum: availabilityValues }) declare availability: (typeof availabilityValues)[number]; + @ApiProperty({ enum: completenessValues }) + declare completeness: (typeof completenessValues)[number]; + @ApiProperty({ enum: freshnessValues }) declare freshness: (typeof freshnessValues)[number]; @ApiProperty({ type: String, format: 'date-time', nullable: true }) declare lastSuccessfulAt: null | string; - @ApiProperty({ type: InvestmentSyncDto }) - declare sync: InvestmentSyncDto; + @ApiProperty({ type: InvestmentAttemptDto }) + declare lastAttempt: InvestmentAttemptDto; + + // Internal resource-source work remains available to existing source owners. + // It is deliberately undecorated and therefore not part of the HTTP schema. + declare sync: { errorCode: null | string; id: null | string; state: 'failed' | 'idle' | 'running' }; } export class InvestmentResourceMetaDto extends InvestmentDependencyDto { diff --git a/src/modules/investments/investment-resource.mapper.spec.ts b/src/modules/investments/investment-resource.mapper.spec.ts index 6ef5727..995be0e 100644 --- a/src/modules/investments/investment-resource.mapper.spec.ts +++ b/src/modules/investments/investment-resource.mapper.spec.ts @@ -1,4 +1,9 @@ +jest.mock('../../prisma/prisma.service', () => ({ + PrismaService: class {}, +})); + import { investmentResourceMeta } from './investment-resource.mapper'; +import { hourlyAttempt } from './tbank-sync/tbank-sync.repository'; describe('investment resource source freshness', () => { afterEach(() => jest.useRealTimers()); @@ -36,6 +41,105 @@ describe('investment resource source freshness', () => { }).freshness, ).toBe('unknown'); }); + + it('keeps confirmed data state independent from a failed hourly attempt', () => { + jest.useFakeTimers().setSystemTime(new Date('2026-09-25T11:30:00Z')); + const meta = investmentResourceMeta({ + resourceKey: 'portfolio/a', + dataRevision: '1', + availability: 'complete', + fetchedAt: new Date('2026-09-25T11:00:00Z'), + freshForMs: 3_600_000, + work: null, + lastAttempt: { + startedAt: '2026-09-25T12:00:00.000Z', + finishedAt: '2026-09-25T12:01:00.000Z', + state: 'partial', + problems: [ + { + section: 'portfolio', + accountId: 'a', + errorCode: 'TBANK_UNAVAILABLE', + }, + ], + }, + }); + + expect(meta).toMatchObject({ + availability: 'available', + completeness: 'complete', + freshness: 'fresh', + lastSuccessfulAt: '2026-09-25T11:00:00.000Z', + lastAttempt: { + state: 'partial', + startedAt: '2026-09-25T12:00:00.000Z', + finishedAt: '2026-09-25T12:01:00.000Z', + }, + }); + }); + + it('projects only the matching hourly section into a resource attempt', () => { + const attempt = hourlyAttempt( + { + startedAt: new Date('2026-09-25T12:00:00Z'), + finishedAt: new Date('2026-09-25T12:01:00Z'), + state: 'partial', + sections: [ + { section: 'portfolio', accountId: 'a', errorCode: 'TBANK_UNAVAILABLE' }, + { section: 'operations', accountId: 'a', errorCode: null }, + { section: 'events', errorCode: 'CALENDAR_UNAVAILABLE' }, + ], + }, + 'portfolio', + 'a', + ); + + const meta = investmentResourceMeta({ + resourceKey: 'portfolio/a', + dataRevision: '1', + availability: 'complete', + fetchedAt: new Date('2026-09-25T11:00:00Z'), + freshForMs: 3_600_000, + work: { id: 'legacy-work', state: 'failed', errorCode: 'LEGACY_FAILURE' }, + lastAttempt: attempt, + }); + + expect(meta).toMatchObject({ + availability: 'available', + completeness: 'complete', + lastAttempt: { + state: 'partial', + startedAt: '2026-09-25T12:00:00.000Z', + finishedAt: '2026-09-25T12:01:00.000Z', + problems: [{ section: 'portfolio', accountId: 'a', errorCode: 'TBANK_UNAVAILABLE' }], + }, + }); + expect(JSON.stringify(meta)).not.toContain('legacy-work'); + }); + + it('projects no confirmed result as incomplete without inventing a successful attempt', () => { + expect( + investmentResourceMeta({ + resourceKey: 'accounts', + dataRevision: '0', + availability: 'missing', + fetchedAt: null, + freshForMs: 3_600_000, + work: { id: 'work', state: 'running', errorCode: null }, + }), + ).toMatchObject({ + availability: 'initial-loading', + completeness: 'incomplete', + freshness: 'unknown', + lastSuccessfulAt: null, + lastAttempt: { + state: 'running', + startedAt: null, + finishedAt: null, + problems: [], + }, + }); + }); }); describe('dependent resource lifecycle', () => { @@ -60,8 +164,8 @@ describe('dependent resource lifecycle', () => { ...input, dependencies: [failed, running], }); - expect(parent.sync).toEqual(running.sync); - expect(parent.availability).toBe('complete'); + expect(parent.lastAttempt.state).toBe('none'); + expect(parent.availability).toBe('available'); const terminal = investmentResourceMeta({ ...input, dependencies: [failed], diff --git a/src/modules/investments/investment-resource.mapper.ts b/src/modules/investments/investment-resource.mapper.ts index ab2e8eb..0a08e65 100644 --- a/src/modules/investments/investment-resource.mapper.ts +++ b/src/modules/investments/investment-resource.mapper.ts @@ -1,41 +1,34 @@ import { InvestmentResourceMetaDto } from './investment-resource.dto'; +type LegacyAvailability = 'available' | 'complete' | 'connection-unavailable' | 'initial-loading' | 'missing' | 'partial' | 'temporarily-unavailable'; +type Work = null | { errorCode: null | string; id: string; state: string }; +type Attempt = NonNullable; + export function investmentResourceMeta(input: { - availability: InvestmentResourceMetaDto['availability']; + availability: LegacyAvailability; coverage?: NonNullable; dataRevision: string; dependencies?: InvestmentResourceMetaDto[]; fetchedAt: Date | null; freshForMs: number; + lastAttempt?: Attempt; resourceKey: string; unavailableReason?: string; - work: null | { errorCode: null | string; id: string; state: string }; + work: Work; }): InvestmentResourceMetaDto { - const ownSync = { - id: input.work?.id ?? null, - state: - input.work?.state === 'running' - ? ('running' as const) - : input.work?.state === 'failed' - ? ('failed' as const) - : ('idle' as const), - errorCode: input.work?.errorCode ?? null, - }; - const states = [ + const ownSync = sourceSync(input.work); + const syncStates = [ ownSync, ...(input.dependencies ?? []) .filter((dependency) => !dependency.unavailableReason) - .map((dependency) => dependency.sync), + .map((dependency) => dependency.sync) + .filter((state): state is typeof ownSync => Boolean(state)), ]; - const sync = - states.find((state) => state.state === 'running') ?? - states.find((state) => state.state === 'failed') ?? + const sourceWork = + syncStates.find((state) => state.state === 'running') ?? + syncStates.find((state) => state.state === 'failed') ?? ownSync; - const sourceFreshness = input.fetchedAt - ? Date.now() - input.fetchedAt.valueOf() < input.freshForMs - ? ('fresh' as const) - : ('stale' as const) - : ('unknown' as const); + const sourceFreshness = freshnessOf(input.fetchedAt, input.freshForMs); const freshness = [ sourceFreshness, ...(input.dependencies ?? []) @@ -57,16 +50,70 @@ export function investmentResourceMeta(input: { ? { unavailableReason: input.unavailableReason } : {}), dataRevision: input.dataRevision, - availability: input.availability, - freshness: freshness.includes('unknown') - ? ('unknown' as const) - : freshness.includes('stale') - ? ('stale' as const) - : ('fresh' as const), + availability: availability(input.availability, input.work), + completeness: + availability(input.availability, input.work) === 'available' + ? ('complete' as const) + : ('incomplete' as const), + freshness: combinedFreshness(freshness), lastSuccessfulAt: timestamps[0] ?? null, - sync, + lastAttempt: input.lastAttempt ?? + attempt(input.work), ...(input.coverage ? { coverage: input.coverage } : {}), ...(input.dependencies ? { dependencies: input.dependencies } : {}), }; - return meta; + return Object.defineProperty(meta, 'sync', { + enumerable: false, + value: sourceWork, + }) as InvestmentResourceMetaDto; +} + +function sourceSync(work: Work) { + let state: 'failed' | 'idle' | 'running' = 'idle'; + if (work?.state === 'running') state = 'running'; + else if (work?.state === 'failed') state = 'failed'; + return { id: work?.id ?? null, state, errorCode: work?.errorCode ?? null }; +} + +function freshnessOf(fetchedAt: Date | null, freshForMs: number) { + if (!fetchedAt) return 'unknown' as const; + return Date.now() - fetchedAt.valueOf() < freshForMs + ? ('fresh' as const) + : ('stale' as const); +} + +function combinedFreshness(values: InvestmentResourceMetaDto['freshness'][]) { + if (values.includes('unknown')) return 'unknown' as const; + if (values.includes('stale')) return 'stale' as const; + return 'fresh' as const; +} + +function attempt(work: Work): Attempt { + const base = { startedAt: null, finishedAt: null }; + if (work?.state === 'running') return { ...base, state: 'running', problems: [] }; + if (work?.state === 'failed') { + return { + ...base, + state: 'failed', + problems: work.errorCode + ? [{ section: 'membership', accountId: null, errorCode: work.errorCode }] + : [], + }; + } + return { ...base, state: 'none', problems: [] }; +} + +function availability(value: LegacyAvailability, work: Work) { + if ( + value === 'available' || + value === 'connection-unavailable' || + value === 'initial-loading' || + value === 'temporarily-unavailable' + ) { + return value; + } + if (value === 'complete') return 'available' as const; + return work?.state === 'running' + ? ('initial-loading' as const) + : ('temporarily-unavailable' as const); } diff --git a/src/modules/investments/operations/application/brokerage-operations.service.ts b/src/modules/investments/operations/application/brokerage-operations.service.ts index e4e3001..5b63076 100644 --- a/src/modules/investments/operations/application/brokerage-operations.service.ts +++ b/src/modules/investments/operations/application/brokerage-operations.service.ts @@ -17,6 +17,10 @@ import { ledgerRange, planInvestmentScreen, } from '../../investment-sync-plan'; +import { + hourlyAttempt, + TbankSyncRepository, +} from '../../tbank-sync/tbank-sync.repository'; import { OperationErrorCode } from '../domain/operation-error-code'; import { confirmedOperationRanges, @@ -59,6 +63,7 @@ export class BrokerageOperationsService { private readonly sync: BrokerageOperationsSyncService, private readonly membership: TbankMembershipService, private readonly config: ConfigService, + private readonly hourlySync: TbankSyncRepository, ) {} async read( @@ -145,6 +150,7 @@ export class BrokerageOperationsService { }); } const saved = await this.repository.readPage(readInput); + const hourly = await this.hourlySync.hourly(userId); const accounts = saved?.connection.accounts ?? []; const targets = accounts.flatMap((account) => from < to @@ -243,6 +249,7 @@ export class BrokerageOperationsService { ? STALE_FRESHNESS_MS : Number.POSITIVE_INFINITY, work: work ?? saved?.membershipWork ?? null, + lastAttempt: hourlyAttempt(hourly, 'operations', accountId === 'all' ? undefined : accountId), dependencies: [membershipMeta, ...metadataDependencies], coverage: { requested: targets.map(serialize), @@ -294,7 +301,7 @@ export class BrokerageOperationsService { const hasNext = Boolean(saved && saved.rows.length > limit); return { data: - meta.availability === 'missing' + meta.completeness === 'incomplete' ? null : { accountId, diff --git a/src/modules/investments/operations/operations.module.ts b/src/modules/investments/operations/operations.module.ts index a91bce8..d62f1ad 100644 --- a/src/modules/investments/operations/operations.module.ts +++ b/src/modules/investments/operations/operations.module.ts @@ -2,13 +2,14 @@ import { Module } from '@nestjs/common'; import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { AuthModule } from '../../auth/auth.module'; +import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module'; import { BrokerageOperationsSyncService } from './application/brokerage-operations-sync.service'; import { BrokerageOperationsService } from './application/brokerage-operations.service'; import { BrokerageOperationsRepository } from './infrastructure/brokerage-operations.repository'; import { BrokerageOperationsController } from './presentation/brokerage-operations.controller'; @Module({ - imports: [AuthModule, TbankModule], + imports: [AuthModule, TbankModule, TbankSyncRepositoryModule], controllers: [BrokerageOperationsController], providers: [ BrokerageOperationsService, diff --git a/src/modules/investments/portfolio/application/forecast.service.ts b/src/modules/investments/portfolio/application/forecast.service.ts index d88d689..fd0811b 100644 --- a/src/modules/investments/portfolio/application/forecast.service.ts +++ b/src/modules/investments/portfolio/application/forecast.service.ts @@ -2,6 +2,7 @@ import { Injectable } from '@nestjs/common'; import type { InvestmentResourceMetaDto } from '../../investment-resource.dto'; +import { add, decimalFromString, formatDecimal } from '../../../../common/decimal'; import { investmentResourceMeta } from '../../investment-resource.mapper'; import { ForecastRepository, @@ -16,6 +17,7 @@ export class ForecastService { async result(input: { candidate?: T; + lastAttempt?: InvestmentResourceMetaDto['lastAttempt']; membershipRevision: number; portfolio: InvestmentResourceMetaDto; revision: string; @@ -72,22 +74,51 @@ export class ForecastService { .map((dependency) => dependency.sync) .find((sync) => sync.state === 'failed'); return { - data: stored ? (stored.payload as T) : null, + data: stored ? filterEvents(stored.payload as T, input.scopeKey) : null, meta: investmentResourceMeta({ resourceKey: input.scopeKey, dataRevision: String(stored?.dataRevision ?? MISSING_DATA_REVISION), availability: stored ? 'complete' : dependencies.some( - (dependency) => dependency.availability !== 'missing', + (dependency) => dependency.completeness === 'complete', ) ? 'partial' : 'missing', fetchedAt: first ? new Date(first) : null, freshForMs: Number.POSITIVE_INFINITY, work: work?.id ? { ...work, id: work.id } : null, + lastAttempt: input.lastAttempt, dependencies, }), }; } } + +function filterEvents(payload: T, scopeKey: string): T { + if (!scopeKey.startsWith('events/')) return payload; + try { + const [, from, to] = JSON.parse(scopeKey.slice('events/'.length)) as [string, string, string]; + const value = payload as T & { + events?: { date: string; projectedAmount?: { currency: string; value: string }; type: string }[]; + incomeTotalsByCurrency?: { currency: string; value: string }[]; + }; + if (!value.events || !value.incomeTotalsByCurrency) return payload; + const events = value.events.filter((event) => event.date >= from && event.date <= to); + const totals = new Map>(); + for (const event of events) { + if (!['coupon', 'dividend'].includes(event.type) || !event.projectedAmount) continue; + const current = totals.get(event.projectedAmount.currency) ?? { value: 0n, scale: 0 }; + totals.set(event.projectedAmount.currency, add(current, decimalFromString(event.projectedAmount.value))); + } + return { + ...value, + events, + incomeTotalsByCurrency: [...totals] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([currency, amount]) => ({ currency, value: formatDecimal(amount) })), + }; + } catch { + return payload; + } +} diff --git a/src/modules/investments/portfolio/application/portfolio.service.spec.ts b/src/modules/investments/portfolio/application/portfolio.service.spec.ts index 8324ca7..433bc75 100644 --- a/src/modules/investments/portfolio/application/portfolio.service.spec.ts +++ b/src/modules/investments/portfolio/application/portfolio.service.spec.ts @@ -26,6 +26,7 @@ describe('persistent portfolio projection', () => { repository as never, { request: jest.fn().mockResolvedValue(null) } as never, { event: jest.fn() } as never, + { hourly: jest.fn().mockResolvedValue(null) } as never, ); access.local.mockResolvedValue({ revision: 'private', @@ -192,7 +193,7 @@ describe('persistent portfolio projection', () => { }); const partial = await service.read('user', 'all'); expect(partial.data).toBeNull(); - expect(partial.meta.availability).toBe('partial'); + expect(partial.meta.availability).toBe('temporarily-unavailable'); const saved = await service.saved('user', 'all'); expect(saved.accounts[0].snapshot?.totalsByCurrency).toEqual( snapshot.totalsByCurrency, diff --git a/src/modules/investments/portfolio/application/portfolio.service.ts b/src/modules/investments/portfolio/application/portfolio.service.ts index eba34dc..2e8a227 100644 --- a/src/modules/investments/portfolio/application/portfolio.service.ts +++ b/src/modules/investments/portfolio/application/portfolio.service.ts @@ -17,6 +17,10 @@ import { tbankSourceTtlMs, } from '../../../../integrations/tbank/tbank.api'; import { investmentResourceMeta } from '../../investment-resource.mapper'; +import { + hourlyAttempt, + TbankSyncRepository, +} from '../../tbank-sync/tbank-sync.repository'; import { portfolioStructure } from '../domain/portfolio-structure'; import { sumMoneyByCurrency } from '../domain/portfolio-totals'; import { @@ -44,12 +48,14 @@ export class PortfolioService { private readonly repository: PortfolioRepository, private readonly membership: TbankMembershipService, private readonly logger: StructuredLogger, + private readonly sync: TbankSyncRepository, ) {} // Local reads never start upstream work; the hourly walk owns refreshes. async saved(userId: string, accountId: string) { const connection = await this.access.local(userId, accountId); const saved = await this.repository.read(userId, connection.revision); + const hourly = await this.sync.hourly(userId); if ( saved?.membershipAt && accountId !== 'all' && @@ -64,8 +70,10 @@ export class PortfolioService { fetchedAt: saved?.membershipAt ?? null, freshForMs: membershipFreshForMs, work: saved?.work.find((work) => work.scopeKey === 'membership') ?? null, + lastAttempt: hourlyAttempt(hourly, 'membership'), }); return { + hourly, membership: membershipMeta, accounts: (saved?.accounts ?? []) .filter( @@ -95,15 +103,16 @@ export class PortfolioService { saved?.work.find( (work) => work.scopeKey === portfolioWorkKey(account.accountId), ) ?? null, + lastAttempt: hourlyAttempt(hourly, 'portfolio', account.accountId), }), })), }; } async read(userId: string, accountId: string) { - const saved = await this.saved(userId, accountId); + const { hourly, ...saved } = await this.saved(userId, accountId); const missing = - saved.membership.availability === 'missing' || + saved.membership.completeness === 'incomplete' || saved.accounts.some((account) => !account.snapshot); const metadata = await this.metadata.savedMany( saved.accounts.flatMap((account) => account.snapshot?.positions ?? []), @@ -206,6 +215,7 @@ export class PortfolioService { ? portfolioFreshForMs : membershipFreshForMs, work: work.id ? { ...work, id: work.id } : null, + lastAttempt: hourlyAttempt(hourly, 'portfolio', accountId === 'all' ? undefined : accountId), dependencies: [ saved.membership, ...saved.accounts.map((account) => account.meta), diff --git a/src/modules/investments/portfolio/infrastructure/forecast.repository.ts b/src/modules/investments/portfolio/infrastructure/forecast.repository.ts index e888736..ca6f52d 100644 --- a/src/modules/investments/portfolio/infrastructure/forecast.repository.ts +++ b/src/modules/investments/portfolio/infrastructure/forecast.repository.ts @@ -51,11 +51,32 @@ export class ForecastRepository { const previous = await tx.tbankForecast.findUnique({ where: { userId_revision_scopeKey: key }, }); + const requestedWindow = eventsWindow(input.scopeKey); + const saved = previous ?? (requestedWindow + ? ( + await tx.tbankForecast.findMany({ + where: { + ...owner, + scopeKey: { startsWith: 'events/' }, + membershipRevision: input.membershipRevision, + }, + }) + ).find((forecast) => { + const window = eventsWindow(forecast.scopeKey); + return ( + window && + window.accountId === requestedWindow.accountId && + window.from <= requestedWindow.to && + requestedWindow.from <= window.to + ); + }) + : null); const fallback = - previous?.membershipRevision === input.membershipRevision - ? previous + saved?.membershipRevision === input.membershipRevision + ? saved : null; if (!input.candidate) return fallback; + if (!previous && fallback) return fallback; const portfolios = await tx.tbankPortfolio.findMany({ where: { ...owner, @@ -154,3 +175,13 @@ export class ForecastRepository { }); } } + +function eventsWindow(scopeKey: string) { + if (!scopeKey.startsWith('events/')) return null; + try { + const [accountId, from, to] = JSON.parse(scopeKey.slice('events/'.length)) as [string, string, string]; + return { accountId, from, to }; + } catch { + return null; + } +} diff --git a/src/modules/investments/portfolio/portfolio.module.ts b/src/modules/investments/portfolio/portfolio.module.ts index 307456c..5b91496 100644 --- a/src/modules/investments/portfolio/portfolio.module.ts +++ b/src/modules/investments/portfolio/portfolio.module.ts @@ -2,6 +2,7 @@ import { Module } from '@nestjs/common'; import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { AuthModule } from '../../auth/auth.module'; +import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module'; import { ForecastService } from './application/forecast.service'; import { PortfolioService } from './application/portfolio.service'; import { ForecastRepository } from './infrastructure/forecast.repository'; @@ -9,7 +10,7 @@ import { PortfolioRepository } from './infrastructure/portfolio.repository'; import { PortfolioController } from './presentation/portfolio.controller'; @Module({ - imports: [AuthModule, TbankModule], + imports: [AuthModule, TbankModule, TbankSyncRepositoryModule], controllers: [PortfolioController], providers: [ PortfolioService, diff --git a/src/modules/investments/tbank-sync/tbank-sync-repository.module.ts b/src/modules/investments/tbank-sync/tbank-sync-repository.module.ts new file mode 100644 index 0000000..fabf453 --- /dev/null +++ b/src/modules/investments/tbank-sync/tbank-sync-repository.module.ts @@ -0,0 +1,9 @@ +import { Module } from '@nestjs/common'; + +import { TbankSyncRepository } from './tbank-sync.repository'; + +@Module({ + providers: [TbankSyncRepository], + exports: [TbankSyncRepository], +}) +export class TbankSyncRepositoryModule {} diff --git a/src/modules/investments/tbank-sync/tbank-sync.dto.ts b/src/modules/investments/tbank-sync/tbank-sync.dto.ts index cbe6bf8..2eb0741 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.dto.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.dto.ts @@ -22,6 +22,7 @@ import { } from 'class-validator'; import { MAX_INSTRUMENT_QUERY_LENGTH, MIN_INSTRUMENT_QUERY_LENGTH } from '../instrument-query'; +import { InvestmentAttemptProblemDto } from '../investment-resource.dto'; import { operationCategories } from '../operations/operations.api'; const MIN_ACCOUNT_ID_LENGTH = 1; @@ -267,6 +268,6 @@ export class HourlySyncStatusDto { @ApiProperty({ type: String, nullable: true, format: 'date-time' }) declare finishedAt: null | string; - @ApiProperty({ type: HourlySyncSectionDto, isArray: true }) - declare sections: HourlySyncSectionDto[]; + @ApiProperty({ type: InvestmentAttemptProblemDto, isArray: true }) + declare problems: InvestmentAttemptProblemDto[]; } diff --git a/src/modules/investments/tbank-sync/tbank-sync.module.ts b/src/modules/investments/tbank-sync/tbank-sync.module.ts index e2df875..44756fb 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.module.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.module.ts @@ -7,8 +7,8 @@ import { OperationsModule } from '../operations/operations.module'; import { PortfolioModule } from '../portfolio/portfolio.module'; import { UpcomingEventsModule } from '../upcoming-events/upcoming-events.module'; import { TbankHourlyScheduler } from './tbank-hourly.scheduler'; +import { TbankSyncRepositoryModule } from './tbank-sync-repository.module'; import { TbankSyncController } from './tbank-sync.controller'; -import { TbankSyncRepository } from './tbank-sync.repository'; import { TbankSyncService } from './tbank-sync.service'; @Module({ imports: [ @@ -18,9 +18,10 @@ import { TbankSyncService } from './tbank-sync.service'; OperationsModule, InstrumentSourcesModule, UpcomingEventsModule, + TbankSyncRepositoryModule, ], controllers: [TbankSyncController], - providers: [TbankSyncService, TbankSyncRepository, TbankHourlyScheduler], - exports: [TbankHourlyScheduler, TbankSyncRepository], + providers: [TbankSyncService, TbankHourlyScheduler], + exports: [TbankHourlyScheduler], }) export class TbankSyncModule {} diff --git a/src/modules/investments/tbank-sync/tbank-sync.repository.ts b/src/modules/investments/tbank-sync/tbank-sync.repository.ts index 0768c95..ae82586 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.repository.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.repository.ts @@ -3,6 +3,7 @@ import { randomUUID } from 'node:crypto'; import type { Prisma } from '../../../../generated/prisma/client'; import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api'; +import type { InvestmentAttemptDto } from '../investment-resource.dto'; import type { InvestmentScreenDto } from './tbank-sync.dto'; import { PrismaService } from '../../../prisma/prisma.service'; @@ -21,6 +22,40 @@ export type HourlyProgress = { result?: 'failed' | 'partial' | 'succeeded'; sections: HourlySection[]; }; +type HourlyAttempt = { + finishedAt: Date | null; + sections: HourlySection[]; + startedAt: Date; + state: 'cancelled' | 'failed' | 'partial' | 'running' | 'succeeded'; +}; + +export function hourlyAttempt( + hourly: HourlyAttempt | null, + section: HourlySection['section'], + accountId?: string, +): InvestmentAttemptDto { + if (!hourly) { + return { state: 'none', startedAt: null, finishedAt: null, problems: [] }; + } + const problems = hourly.sections + .filter( + (entry) => + entry.section === section && + (accountId === undefined || entry.accountId === accountId), + ) + .map((entry) => ({ + section: entry.section, + accountId: entry.accountId ?? null, + errorCode: entry.errorCode, + })); + return { + state: hourly.state === 'cancelled' ? 'failed' : hourly.state, + startedAt: hourly.startedAt.toISOString(), + finishedAt: hourly.finishedAt?.toISOString() ?? null, + problems, + }; +} + @Injectable() export class TbankSyncRepository { constructor(private readonly prisma: PrismaService) {} diff --git a/src/modules/investments/upcoming-events/upcoming-events.module.ts b/src/modules/investments/upcoming-events/upcoming-events.module.ts index ed66431..ab7ae36 100644 --- a/src/modules/investments/upcoming-events/upcoming-events.module.ts +++ b/src/modules/investments/upcoming-events/upcoming-events.module.ts @@ -4,11 +4,12 @@ import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { AuthModule } from '../../auth/auth.module'; import { InstrumentSourcesModule } from '../instruments/instrument-sources.module'; import { PortfolioModule } from '../portfolio/portfolio.module'; +import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module'; import { UpcomingEventsController } from './upcoming-events.controller'; import { UpcomingEventsService } from './upcoming-events.service'; @Module({ - imports: [AuthModule, TbankModule, InstrumentSourcesModule, PortfolioModule], + imports: [AuthModule, TbankModule, InstrumentSourcesModule, PortfolioModule, TbankSyncRepositoryModule], controllers: [UpcomingEventsController], providers: [UpcomingEventsService], exports: [UpcomingEventsService], diff --git a/src/modules/investments/upcoming-events/upcoming-events.service.ts b/src/modules/investments/upcoming-events/upcoming-events.service.ts index 937c231..5393afc 100644 --- a/src/modules/investments/upcoming-events/upcoming-events.service.ts +++ b/src/modules/investments/upcoming-events/upcoming-events.service.ts @@ -20,6 +20,10 @@ import { type PortfolioPositionDto, PortfolioService, } from '../portfolio/portfolio.api'; +import { + hourlyAttempt, + TbankSyncRepository, +} from '../tbank-sync/tbank-sync.repository'; import { bondEvent, dividendEvent } from './upcoming-event.mapper'; import { UpcomingEventDto, @@ -43,6 +47,7 @@ export class UpcomingEventsService { private readonly sources: CatalogEnrichmentService, private readonly forecasts: ForecastService, private readonly metadata: InstrumentMetadataService, + private readonly sync: TbankSyncRepository, ) {} // Read endpoints stay local; callers without a saved portfolio receive unknown. @@ -54,6 +59,7 @@ export class UpcomingEventsService { const window = dateWindow(query); const connection = await this.access.local(userId, accountId); const portfolio = await this.portfolios.read(userId, accountId); + const hourly = await this.sync.hourly(userId); const events: UpcomingEventDto[] = []; let ready = portfolio.data !== null; const positions = portfolio.data?.positions.filter(supported) ?? []; @@ -61,7 +67,7 @@ export class UpcomingEventsService { userId, positions, window, - query.sync === 'observe', + true, ); for (const position of positions) { const generic = metadata.get(metadataKey(position))?.data; @@ -98,7 +104,7 @@ export class UpcomingEventsService { source.input.uid === uid && source.input.kind === 'detail', )?.instrument || generic; const ticker = position.ticker || instrument?.ticker; - if (!calendar || calendar.meta.availability !== 'complete' || !ticker) { + if (!calendar || calendar.meta.completeness !== 'complete' || !ticker) { ready = false; continue; } @@ -157,6 +163,11 @@ export class UpcomingEventsService { membershipRevision: connection.membershipRevision, scopeKey: `events/${JSON.stringify([accountId, window.from, window.to])}`, portfolio: portfolio.meta, + lastAttempt: hourlyAttempt( + hourly, + 'events', + accountId === 'all' ? undefined : accountId, + ), sources, candidate, }); @@ -168,8 +179,12 @@ export class UpcomingEventsService { // Triggers online refreshes of the account's event sources and returns the // launched work ids for the hourly walk to observe. - async refresh(userId: string, accountId: string): Promise { - const window = dateWindow({}); + async refresh( + userId: string, + accountId: string, + query: UpcomingEventsQueryDto = {}, + ): Promise { + const window = dateWindow(query); await this.access.local(userId, accountId); const portfolio = await this.portfolios.read(userId, accountId); const positions = portfolio.data?.positions.filter(supported) ?? [];