feat: reconcile hourly investment resource states #50

Merged
ksv741 merged 7 commits from codex/reconcile-hourly-investment-sync into develop 2026-10-09 23:10:50 +03:00
34 changed files with 571 additions and 176 deletions
-15
View File
@@ -32,11 +32,6 @@
"count": 1 "count": 1
} }
}, },
"src/modules/investments/brokerage-accounts/brokerage-accounts.service.ts": {
"sonarjs/no-nested-conditional": {
"count": 1
}
},
"src/modules/investments/income/income-analytics.service.ts": { "src/modules/investments/income/income-analytics.service.ts": {
"sonarjs/no-nested-template-literals": { "sonarjs/no-nested-template-literals": {
"count": 1 "count": 1
@@ -99,11 +94,6 @@
"count": 1 "count": 1
} }
}, },
"src/modules/investments/investment-resource.mapper.ts": {
"sonarjs/no-nested-conditional": {
"count": 3
}
},
"src/modules/investments/investment-sync-plan.ts": { "src/modules/investments/investment-sync-plan.ts": {
"sonarjs/cognitive-complexity": { "sonarjs/cognitive-complexity": {
"count": 1 "count": 1
@@ -146,11 +136,6 @@
"count": 1 "count": 1
} }
}, },
"src/modules/investments/portfolio/application/portfolio.service.ts": {
"sonarjs/no-nested-conditional": {
"count": 1
}
},
"src/modules/investments/tbank-sync/tbank-screen.ts": { "src/modules/investments/tbank-sync/tbank-screen.ts": {
"sonarjs/cognitive-complexity": { "sonarjs/cognitive-complexity": {
"count": 1 "count": 1
@@ -3,12 +3,12 @@ import { Module } from '@nestjs/common';
import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { TbankModule } from '../../../integrations/tbank/tbank.module';
import { AuthModule } from '../../auth/auth.module'; import { AuthModule } from '../../auth/auth.module';
import { PortfolioModule } from '../portfolio/portfolio.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 { BrokerageAccountsController } from './brokerage-accounts.controller';
import { BrokerageAccountsService } from './brokerage-accounts.service'; import { BrokerageAccountsService } from './brokerage-accounts.service';
@Module({ @Module({
imports: [AuthModule, TbankModule, PortfolioModule, TbankSyncModule], imports: [AuthModule, TbankModule, PortfolioModule, TbankSyncRepositoryModule],
controllers: [BrokerageAccountsController], controllers: [BrokerageAccountsController],
providers: [BrokerageAccountsService], providers: [BrokerageAccountsService],
exports: [BrokerageAccountsService], exports: [BrokerageAccountsService],
@@ -24,7 +24,7 @@ describe('saved brokerage account projection', () => {
it('distinguishes unknown membership from a confirmed empty list', async () => { it('distinguishes unknown membership from a confirmed empty list', async () => {
portfolio.saved.mockResolvedValueOnce({ portfolio.saved.mockResolvedValueOnce({
membership: { ...meta, availability: 'missing' }, membership: { ...meta, completeness: 'incomplete' },
accounts: [], accounts: [],
}); });
expect((await service.list('owner')).data).toBeNull(); expect((await service.list('owner')).data).toBeNull();
@@ -66,6 +66,39 @@ describe('saved brokerage account projection', () => {
bondPositionCount: 1, bondPositionCount: 1,
}); });
expect(response.data?.[1].portfolioSummary).toBeNull(); 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');
}); });
}); });
@@ -16,7 +16,7 @@ export class BrokerageAccountsService {
async list(userId: string): Promise<BrokerageAccountsResourceDto> { async list(userId: string): Promise<BrokerageAccountsResourceDto> {
const saved = await this.portfolio.saved(userId, 'all'); const saved = await this.portfolio.saved(userId, 'all');
const data: BrokerageAccountDto[] | null = const data: BrokerageAccountDto[] | null =
saved.membership.availability === 'missing' saved.membership.completeness === 'incomplete'
? null ? null
: saved.accounts.map((account) => ({ : saved.accounts.map((account) => ({
id: account.id, id: account.id,
@@ -55,16 +55,12 @@ export class BrokerageAccountsService {
availability: availability:
data === null data === null
? 'missing' ? 'missing'
: saved.accounts.some((account) => !account.snapshot) : 'complete',
? 'partial'
: 'complete',
fetchedAt: saved.membership.lastSuccessfulAt fetchedAt: saved.membership.lastSuccessfulAt
? new Date(saved.membership.lastSuccessfulAt) ? new Date(saved.membership.lastSuccessfulAt)
: null, : null,
freshForMs: 3_600_000, freshForMs: 3_600_000,
work: saved.membership.sync.id work: null,
? { ...saved.membership.sync, id: saved.membership.sync.id }
: null,
dependencies: saved.accounts.map((account) => account.meta), dependencies: saved.accounts.map((account) => account.meta),
}), }),
sync: hourly sync: hourly
@@ -72,7 +68,10 @@ export class BrokerageAccountsService {
state: hourly.state, state: hourly.state,
startedAt: hourly.startedAt.toISOString(), startedAt: hourly.startedAt.toISOString(),
finishedAt: hourly.finishedAt?.toISOString() ?? null, finishedAt: hourly.finishedAt?.toISOString() ?? null,
sections: hourly.sections, problems: hourly.sections.map((section) => ({
...section,
accountId: section.accountId ?? null,
})),
} }
: null, : null,
}; };
@@ -48,7 +48,7 @@ describe('IncomeAnalyticsService', () => {
expect.objectContaining({ role: 'previous' }), expect.objectContaining({ role: 'previous' }),
expect.objectContaining({ expect.objectContaining({
role: 'year', role: 'year',
availability: 'missing', availability: 'temporarily-unavailable',
unavailableReason: 'CALENDAR_OR_ACCOUNT_HISTORY', unavailableReason: 'CALENDAR_OR_ACCOUNT_HISTORY',
}), }),
]); ]);
@@ -119,7 +119,10 @@ export class IncomeAnalyticsService {
...investmentResourceMeta({ ...investmentResourceMeta({
resourceKey: `income/${accountId}/${plan.selected.from}/${plan.selected.to}`, resourceKey: `income/${accountId}/${plan.selected.from}/${plan.selected.to}`,
dataRevision: current.meta.dataRevision, dataRevision: current.meta.dataRevision,
availability: current.meta.availability, availability:
current.items === null
? 'temporarily-unavailable'
: 'available',
fetchedAt: current.meta.lastSuccessfulAt fetchedAt: current.meta.lastSuccessfulAt
? new Date(current.meta.lastSuccessfulAt) ? new Date(current.meta.lastSuccessfulAt)
: null, : null,
@@ -29,6 +29,7 @@ import { timestamp } from './instrument-market';
const REPLACED_WINDOW_COUNT = 1; const REPLACED_WINDOW_COUNT = 1;
const NEXT_WINDOW_OFFSET = 1; const NEXT_WINDOW_OFFSET = 1;
const NO_CANDLE_GAPS = 0;
type CandleInput = { type CandleInput = {
instrumentType: 'bond' | 'share'; instrumentType: 'bond' | 'share';
@@ -106,7 +107,7 @@ export class CandleService {
true, true,
['detail'], ['detail'],
); );
if (detail.meta.availability === 'complete' && !detail.instrument) { if (detail.meta.completeness === 'complete' && !detail.instrument) {
throw new NotFoundException(); throw new NotFoundException();
} }
const now = new Date(); const now = new Date();
@@ -114,8 +115,10 @@ export class CandleService {
? rangeFor(input, detail.instrument, now) ? rangeFor(input, detail.instrument, now)
: null; : null;
const saved = range ? await this.repository.read(input.uid, range) : null; const saved = range ? await this.repository.read(input.uid, range) : null;
const gaps = range && saved ? candleGaps(range, saved.coverage, now) : []; const hasCompleteRange = Boolean(
if (!observe && (!saved || gaps.length)) { range && saved && candleGaps(range, saved.coverage, now).length === NO_CANDLE_GAPS,
);
if (!observe && (!saved || !hasCompleteRange)) {
await this.request(userId, revision, input, now); await this.request(userId, revision, input, now);
} }
const work = await this.repository.work(userId, revision, scope(input)); const work = await this.repository.work(userId, revision, scope(input));
@@ -121,7 +121,7 @@ export class CatalogDiscoveryService {
availability: availability:
discovery.data === null discovery.data === null
? 'missing' ? 'missing'
: sources.every((source) => source.meta.availability === 'complete') : sources.every((source) => source.meta.completeness === 'complete')
? 'complete' ? 'complete'
: 'partial', : 'partial',
fetchedAt: discovery.meta.lastSuccessfulAt fetchedAt: discovery.meta.lastSuccessfulAt
@@ -173,7 +173,7 @@ export class CatalogDiscoveryService {
['detail'], ['detail'],
); );
const source = sources[0]; const source = sources[0];
if (source.meta.availability === 'complete' && !source.instrument) { if (source.meta.completeness === 'complete' && !source.instrument) {
throw new NotFoundException(); throw new NotFoundException();
} }
const instrument = source.instrument; const instrument = source.instrument;
@@ -288,7 +288,7 @@ export class CatalogEnrichmentService {
kinds, kinds,
); );
const first = sources[0]; const first = sources[0];
if (first.meta.availability === 'complete' && !first.instrument) { if (first.meta.completeness === 'complete' && !first.instrument) {
throw new NotFoundException(); throw new NotFoundException();
} }
const values = sources.reduce<CatalogPayload>( const values = sources.reduce<CatalogPayload>(
@@ -297,7 +297,7 @@ export class CatalogEnrichmentService {
); );
const dependencies = sources.map((source) => source.meta); const dependencies = sources.map((source) => source.meta);
const complete = dependencies.every( const complete = dependencies.every(
(dependency) => dependency.availability === 'complete', (dependency) => dependency.completeness === 'complete',
); );
const meta = investmentResourceMeta({ const meta = investmentResourceMeta({
resourceKey: `${instrumentType}/${section}/${uid}`, resourceKey: `${instrumentType}/${section}/${uid}`,
@@ -322,7 +322,7 @@ export class CatalogEnrichmentService {
unavailable: sources unavailable: sources
.filter( .filter(
(source) => (source) =>
source.meta.availability !== 'complete' || source.meta.completeness !== 'complete' ||
source.meta.sync.state === 'failed', source.meta.sync.state === 'failed',
) )
.map((source) => source.input.kind), .map((source) => source.input.kind),
@@ -97,7 +97,7 @@ export class BondIncomeService {
['detail', 'coupons'], ['detail', 'coupons'],
{ from, to }, { from, to },
); );
if (sources[0].meta.availability === 'complete' && !sources[0].instrument) { if (sources[0].meta.completeness === 'complete' && !sources[0].instrument) {
throw new NotFoundException(); throw new NotFoundException();
} }
const events = sources[1].payload?.coupons; const events = sources[1].payload?.coupons;
@@ -137,7 +137,7 @@ export class BondIncomeService {
projected.observedAt, projected.observedAt,
true, true,
), ),
meta: { ...result.meta, availability: 'partial' }, meta: { ...result.meta, completeness: 'incomplete' },
}; };
} }
} }
@@ -66,7 +66,7 @@ export class StockIncomeService {
['detail', 'dividends'], ['detail', 'dividends'],
{ from, to }, { from, to },
); );
if (sources[0].meta.availability === 'complete' && !sources[0].instrument) { if (sources[0].meta.completeness === 'complete' && !sources[0].instrument) {
throw new NotFoundException(); throw new NotFoundException();
} }
const events = sources[1].payload?.dividends; const events = sources[1].payload?.dividends;
@@ -106,7 +106,7 @@ export class StockIncomeService {
projected.observedAt, projected.observedAt,
true, true,
), ),
meta: { ...result.meta, availability: 'partial' }, meta: { ...result.meta, completeness: 'incomplete' },
}; };
} }
} }
@@ -1,9 +1,21 @@
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
import { IsIn, IsOptional } from 'class-validator'; 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 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 { export class InvestmentReadQueryDto {
@ApiPropertyOptional({ enum: ['observe'] }) @ApiPropertyOptional({ enum: ['observe'] })
@@ -12,17 +24,31 @@ export class InvestmentReadQueryDto {
declare sync?: 'observe'; declare sync?: 'observe';
} }
export class InvestmentSyncDto { export class InvestmentAttemptProblemDto {
@ApiProperty({ type: String, nullable: true }) @ApiProperty({ enum: ['membership', 'portfolio', 'operations', 'events'] })
declare id: null | string; declare section: 'events' | 'membership' | 'operations' | 'portfolio';
@ApiProperty({ enum: syncStateValues }) @ApiProperty({ type: String, nullable: true })
declare state: (typeof syncStateValues)[number]; declare accountId: null | string;
@ApiProperty({ type: String, nullable: true }) @ApiProperty({ type: String, nullable: true })
declare errorCode: null | string; 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 { export class InvestmentRangeDto {
@ApiPropertyOptional() @ApiPropertyOptional()
accountId?: string; accountId?: string;
@@ -73,14 +99,21 @@ export class InvestmentDependencyDto {
@ApiProperty({ enum: availabilityValues }) @ApiProperty({ enum: availabilityValues })
declare availability: (typeof availabilityValues)[number]; declare availability: (typeof availabilityValues)[number];
@ApiProperty({ enum: completenessValues })
declare completeness: (typeof completenessValues)[number];
@ApiProperty({ enum: freshnessValues }) @ApiProperty({ enum: freshnessValues })
declare freshness: (typeof freshnessValues)[number]; declare freshness: (typeof freshnessValues)[number];
@ApiProperty({ type: String, format: 'date-time', nullable: true }) @ApiProperty({ type: String, format: 'date-time', nullable: true })
declare lastSuccessfulAt: null | string; declare lastSuccessfulAt: null | string;
@ApiProperty({ type: InvestmentSyncDto }) @ApiProperty({ type: InvestmentAttemptDto })
declare sync: InvestmentSyncDto; 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 { export class InvestmentResourceMetaDto extends InvestmentDependencyDto {
@@ -1,4 +1,9 @@
jest.mock('../../prisma/prisma.service', () => ({
PrismaService: class {},
}));
import { investmentResourceMeta } from './investment-resource.mapper'; import { investmentResourceMeta } from './investment-resource.mapper';
import { hourlyAttempt } from './tbank-sync/tbank-sync.repository';
describe('investment resource source freshness', () => { describe('investment resource source freshness', () => {
afterEach(() => jest.useRealTimers()); afterEach(() => jest.useRealTimers());
@@ -36,6 +41,105 @@ describe('investment resource source freshness', () => {
}).freshness, }).freshness,
).toBe('unknown'); ).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', () => { describe('dependent resource lifecycle', () => {
@@ -60,8 +164,8 @@ describe('dependent resource lifecycle', () => {
...input, ...input,
dependencies: [failed, running], dependencies: [failed, running],
}); });
expect(parent.sync).toEqual(running.sync); expect(parent.lastAttempt.state).toBe('none');
expect(parent.availability).toBe('complete'); expect(parent.availability).toBe('available');
const terminal = investmentResourceMeta({ const terminal = investmentResourceMeta({
...input, ...input,
dependencies: [failed], dependencies: [failed],
@@ -1,41 +1,34 @@
import { InvestmentResourceMetaDto } from './investment-resource.dto'; 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<InvestmentResourceMetaDto['lastAttempt']>;
export function investmentResourceMeta(input: { export function investmentResourceMeta(input: {
availability: InvestmentResourceMetaDto['availability']; availability: LegacyAvailability;
coverage?: NonNullable<InvestmentResourceMetaDto['coverage']>; coverage?: NonNullable<InvestmentResourceMetaDto['coverage']>;
dataRevision: string; dataRevision: string;
dependencies?: InvestmentResourceMetaDto[]; dependencies?: InvestmentResourceMetaDto[];
fetchedAt: Date | null; fetchedAt: Date | null;
freshForMs: number; freshForMs: number;
lastAttempt?: Attempt;
resourceKey: string; resourceKey: string;
unavailableReason?: string; unavailableReason?: string;
work: null | { errorCode: null | string; id: string; state: string }; work: Work;
}): InvestmentResourceMetaDto { }): InvestmentResourceMetaDto {
const ownSync = { const ownSync = sourceSync(input.work);
id: input.work?.id ?? null, const syncStates = [
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 = [
ownSync, ownSync,
...(input.dependencies ?? []) ...(input.dependencies ?? [])
.filter((dependency) => !dependency.unavailableReason) .filter((dependency) => !dependency.unavailableReason)
.map((dependency) => dependency.sync), .map((dependency) => dependency.sync)
.filter((state): state is typeof ownSync => Boolean(state)),
]; ];
const sync = const sourceWork =
states.find((state) => state.state === 'running') ?? syncStates.find((state) => state.state === 'running') ??
states.find((state) => state.state === 'failed') ?? syncStates.find((state) => state.state === 'failed') ??
ownSync; ownSync;
const sourceFreshness = input.fetchedAt const sourceFreshness = freshnessOf(input.fetchedAt, input.freshForMs);
? Date.now() - input.fetchedAt.valueOf() < input.freshForMs
? ('fresh' as const)
: ('stale' as const)
: ('unknown' as const);
const freshness = [ const freshness = [
sourceFreshness, sourceFreshness,
...(input.dependencies ?? []) ...(input.dependencies ?? [])
@@ -57,16 +50,75 @@ export function investmentResourceMeta(input: {
? { unavailableReason: input.unavailableReason } ? { unavailableReason: input.unavailableReason }
: {}), : {}),
dataRevision: input.dataRevision, dataRevision: input.dataRevision,
availability: input.availability, availability: availability(input.availability, input.work),
freshness: freshness.includes('unknown') completeness:
? ('unknown' as const) availability(input.availability, input.work) === 'available'
: freshness.includes('stale') ? ('complete' as const)
? ('stale' as const) : ('incomplete' as const),
: ('fresh' as const), freshness: combinedFreshness(freshness),
lastSuccessfulAt: timestamps[0] ?? null, lastSuccessfulAt: timestamps[0] ?? null,
sync, lastAttempt: input.lastAttempt ??
attempt(input.work),
...(input.coverage ? { coverage: input.coverage } : {}), ...(input.coverage ? { coverage: input.coverage } : {}),
...(input.dependencies ? { dependencies: input.dependencies } : {}), ...(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;
if (value === 'partial') {
return work?.state === 'running'
? ('initial-loading' as const)
: ('temporarily-unavailable' as const);
}
return work?.state === 'running'
? ('initial-loading' as const)
: ('temporarily-unavailable' as const);
} }
@@ -17,6 +17,10 @@ import {
ledgerRange, ledgerRange,
planInvestmentScreen, planInvestmentScreen,
} from '../../investment-sync-plan'; } from '../../investment-sync-plan';
import {
hourlyAttempt,
TbankSyncRepository,
} from '../../tbank-sync/tbank-sync.repository';
import { OperationErrorCode } from '../domain/operation-error-code'; import { OperationErrorCode } from '../domain/operation-error-code';
import { import {
confirmedOperationRanges, confirmedOperationRanges,
@@ -59,6 +63,7 @@ export class BrokerageOperationsService {
private readonly sync: BrokerageOperationsSyncService, private readonly sync: BrokerageOperationsSyncService,
private readonly membership: TbankMembershipService, private readonly membership: TbankMembershipService,
private readonly config: ConfigService, private readonly config: ConfigService,
private readonly hourlySync: TbankSyncRepository,
) {} ) {}
async read( async read(
@@ -145,6 +150,7 @@ export class BrokerageOperationsService {
}); });
} }
const saved = await this.repository.readPage(readInput); const saved = await this.repository.readPage(readInput);
const hourly = await this.hourlySync.hourly(userId);
const accounts = saved?.connection.accounts ?? []; const accounts = saved?.connection.accounts ?? [];
const targets = accounts.flatMap((account) => const targets = accounts.flatMap((account) =>
from < to from < to
@@ -243,6 +249,7 @@ export class BrokerageOperationsService {
? STALE_FRESHNESS_MS ? STALE_FRESHNESS_MS
: Number.POSITIVE_INFINITY, : Number.POSITIVE_INFINITY,
work: work ?? saved?.membershipWork ?? null, work: work ?? saved?.membershipWork ?? null,
lastAttempt: hourlyAttempt(hourly, 'operations', accountId === 'all' ? undefined : accountId),
dependencies: [membershipMeta, ...metadataDependencies], dependencies: [membershipMeta, ...metadataDependencies],
coverage: { coverage: {
requested: targets.map(serialize), requested: targets.map(serialize),
@@ -294,7 +301,7 @@ export class BrokerageOperationsService {
const hasNext = Boolean(saved && saved.rows.length > limit); const hasNext = Boolean(saved && saved.rows.length > limit);
return { return {
data: data:
meta.availability === 'missing' meta.completeness === 'incomplete'
? null ? null
: { : {
accountId, accountId,
@@ -2,13 +2,14 @@ import { Module } from '@nestjs/common';
import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { TbankModule } from '../../../integrations/tbank/tbank.module';
import { AuthModule } from '../../auth/auth.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 { BrokerageOperationsSyncService } from './application/brokerage-operations-sync.service';
import { BrokerageOperationsService } from './application/brokerage-operations.service'; import { BrokerageOperationsService } from './application/brokerage-operations.service';
import { BrokerageOperationsRepository } from './infrastructure/brokerage-operations.repository'; import { BrokerageOperationsRepository } from './infrastructure/brokerage-operations.repository';
import { BrokerageOperationsController } from './presentation/brokerage-operations.controller'; import { BrokerageOperationsController } from './presentation/brokerage-operations.controller';
@Module({ @Module({
imports: [AuthModule, TbankModule], imports: [AuthModule, TbankModule, TbankSyncRepositoryModule],
controllers: [BrokerageOperationsController], controllers: [BrokerageOperationsController],
providers: [ providers: [
BrokerageOperationsService, BrokerageOperationsService,
@@ -2,6 +2,7 @@ import { Injectable } from '@nestjs/common';
import type { InvestmentResourceMetaDto } from '../../investment-resource.dto'; import type { InvestmentResourceMetaDto } from '../../investment-resource.dto';
import { add, decimalFromString, formatDecimal } from '../../../../common/decimal';
import { investmentResourceMeta } from '../../investment-resource.mapper'; import { investmentResourceMeta } from '../../investment-resource.mapper';
import { import {
ForecastRepository, ForecastRepository,
@@ -16,6 +17,7 @@ export class ForecastService {
async result<T extends object>(input: { async result<T extends object>(input: {
candidate?: T; candidate?: T;
lastAttempt?: InvestmentResourceMetaDto['lastAttempt'];
membershipRevision: number; membershipRevision: number;
portfolio: InvestmentResourceMetaDto; portfolio: InvestmentResourceMetaDto;
revision: string; revision: string;
@@ -72,22 +74,51 @@ export class ForecastService {
.map((dependency) => dependency.sync) .map((dependency) => dependency.sync)
.find((sync) => sync.state === 'failed'); .find((sync) => sync.state === 'failed');
return { return {
data: stored ? (stored.payload as T) : null, data: stored ? filterEvents(stored.payload as T, input.scopeKey) : null,
meta: investmentResourceMeta({ meta: investmentResourceMeta({
resourceKey: input.scopeKey, resourceKey: input.scopeKey,
dataRevision: String(stored?.dataRevision ?? MISSING_DATA_REVISION), dataRevision: String(stored?.dataRevision ?? MISSING_DATA_REVISION),
availability: stored availability: stored
? 'complete' ? 'complete'
: dependencies.some( : dependencies.some(
(dependency) => dependency.availability !== 'missing', (dependency) => dependency.completeness === 'complete',
) )
? 'partial' ? 'partial'
: 'missing', : 'missing',
fetchedAt: first ? new Date(first) : null, fetchedAt: first ? new Date(first) : null,
freshForMs: Number.POSITIVE_INFINITY, freshForMs: Number.POSITIVE_INFINITY,
work: work?.id ? { ...work, id: work.id } : null, work: work?.id ? { ...work, id: work.id } : null,
lastAttempt: input.lastAttempt,
dependencies, dependencies,
}), }),
}; };
} }
} }
function filterEvents<T extends object>(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<string, ReturnType<typeof decimalFromString>>();
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;
}
}
@@ -26,6 +26,7 @@ describe('persistent portfolio projection', () => {
repository as never, repository as never,
{ request: jest.fn().mockResolvedValue(null) } as never, { request: jest.fn().mockResolvedValue(null) } as never,
{ event: jest.fn() } as never, { event: jest.fn() } as never,
{ hourly: jest.fn().mockResolvedValue(null) } as never,
); );
access.local.mockResolvedValue({ access.local.mockResolvedValue({
revision: 'private', revision: 'private',
@@ -192,7 +193,8 @@ describe('persistent portfolio projection', () => {
}); });
const partial = await service.read('user', 'all'); const partial = await service.read('user', 'all');
expect(partial.data).toBeNull(); expect(partial.data).toBeNull();
expect(partial.meta.availability).toBe('partial'); expect(partial.meta.availability).toBe('temporarily-unavailable');
expect(partial.meta.completeness).toBe('incomplete');
const saved = await service.saved('user', 'all'); const saved = await service.saved('user', 'all');
expect(saved.accounts[0].snapshot?.totalsByCurrency).toEqual( expect(saved.accounts[0].snapshot?.totalsByCurrency).toEqual(
snapshot.totalsByCurrency, snapshot.totalsByCurrency,
@@ -17,6 +17,10 @@ import {
tbankSourceTtlMs, tbankSourceTtlMs,
} from '../../../../integrations/tbank/tbank.api'; } from '../../../../integrations/tbank/tbank.api';
import { investmentResourceMeta } from '../../investment-resource.mapper'; import { investmentResourceMeta } from '../../investment-resource.mapper';
import {
hourlyAttempt,
TbankSyncRepository,
} from '../../tbank-sync/tbank-sync.repository';
import { portfolioStructure } from '../domain/portfolio-structure'; import { portfolioStructure } from '../domain/portfolio-structure';
import { sumMoneyByCurrency } from '../domain/portfolio-totals'; import { sumMoneyByCurrency } from '../domain/portfolio-totals';
import { import {
@@ -28,6 +32,7 @@ import { mapPortfolio } from '../infrastructure/tbank-portfolio.mapper';
import { PortfolioDto } from '../presentation/portfolio.dto'; import { PortfolioDto } from '../presentation/portfolio.dto';
const MISSING_DATA_REVISION = 0; const MISSING_DATA_REVISION = 0;
const NO_ACCOUNTS = 0;
const membershipFreshForMs = tbankSourceTtlMs.membership; const membershipFreshForMs = tbankSourceTtlMs.membership;
@@ -44,12 +49,14 @@ export class PortfolioService {
private readonly repository: PortfolioRepository, private readonly repository: PortfolioRepository,
private readonly membership: TbankMembershipService, private readonly membership: TbankMembershipService,
private readonly logger: StructuredLogger, private readonly logger: StructuredLogger,
private readonly sync: TbankSyncRepository,
) {} ) {}
// Local reads never start upstream work; the hourly walk owns refreshes. // Local reads never start upstream work; the hourly walk owns refreshes.
async saved(userId: string, accountId: string) { async saved(userId: string, accountId: string) {
const connection = await this.access.local(userId, accountId); const connection = await this.access.local(userId, accountId);
const saved = await this.repository.read(userId, connection.revision); const saved = await this.repository.read(userId, connection.revision);
const hourly = await this.sync.hourly(userId);
if ( if (
saved?.membershipAt && saved?.membershipAt &&
accountId !== 'all' && accountId !== 'all' &&
@@ -64,8 +71,10 @@ export class PortfolioService {
fetchedAt: saved?.membershipAt ?? null, fetchedAt: saved?.membershipAt ?? null,
freshForMs: membershipFreshForMs, freshForMs: membershipFreshForMs,
work: saved?.work.find((work) => work.scopeKey === 'membership') ?? null, work: saved?.work.find((work) => work.scopeKey === 'membership') ?? null,
lastAttempt: hourlyAttempt(hourly, 'membership'),
}); });
return { return {
hourly,
membership: membershipMeta, membership: membershipMeta,
accounts: (saved?.accounts ?? []) accounts: (saved?.accounts ?? [])
.filter( .filter(
@@ -95,16 +104,21 @@ export class PortfolioService {
saved?.work.find( saved?.work.find(
(work) => work.scopeKey === portfolioWorkKey(account.accountId), (work) => work.scopeKey === portfolioWorkKey(account.accountId),
) ?? null, ) ?? null,
lastAttempt: hourlyAttempt(hourly, 'portfolio', account.accountId),
}), }),
})), })),
}; };
} }
async read(userId: string, accountId: string) { async read(userId: string, accountId: string) {
const saved = await this.saved(userId, accountId); const { hourly, ...saved } = await this.saved(userId, accountId);
const missing = const missing =
saved.membership.availability === 'missing' || saved.membership.completeness === 'incomplete' ||
saved.accounts.some((account) => !account.snapshot); (saved.accounts.length > NO_ACCOUNTS && saved.accounts.some((account) => !account.snapshot));
const available =
!missing &&
(saved.membership.availability === 'available' ||
saved.accounts.some((account) => account.snapshot));
const metadata = await this.metadata.savedMany( const metadata = await this.metadata.savedMany(
saved.accounts.flatMap((account) => account.snapshot?.positions ?? []), saved.accounts.flatMap((account) => account.snapshot?.positions ?? []),
); );
@@ -153,6 +167,11 @@ export class PortfolioService {
] ]
: [], : [],
); );
let availability: 'available' | 'complete' | 'missing' | 'partial';
if (available) availability = 'available';
else if (!missing) availability = 'complete';
else if (portfolios.length > NO_ACCOUNTS) availability = 'partial';
else availability = 'missing';
const lastSuccessfulAt = const lastSuccessfulAt =
// eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings. // eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings.
portfolios.map((portfolio) => portfolio.lastSuccessfulAt).sort()[0] ?? portfolios.map((portfolio) => portfolio.lastSuccessfulAt).sort()[0] ??
@@ -196,16 +215,13 @@ export class PortfolioService {
]), ]),
metadataDependencies.map((dependency) => dependency.dataRevision), metadataDependencies.map((dependency) => dependency.dataRevision),
]), ]),
availability: missing availability,
? portfolios.length
? 'partial'
: 'missing'
: 'complete',
fetchedAt: lastSuccessfulAt ? new Date(lastSuccessfulAt) : null, fetchedAt: lastSuccessfulAt ? new Date(lastSuccessfulAt) : null,
freshForMs: portfolios.length freshForMs: portfolios.length
? portfolioFreshForMs ? portfolioFreshForMs
: membershipFreshForMs, : membershipFreshForMs,
work: work.id ? { ...work, id: work.id } : null, work: work.id ? { ...work, id: work.id } : null,
lastAttempt: hourlyAttempt(hourly, 'portfolio', accountId === 'all' ? undefined : accountId),
dependencies: [ dependencies: [
saved.membership, saved.membership,
...saved.accounts.map((account) => account.meta), ...saved.accounts.map((account) => account.meta),
@@ -51,11 +51,32 @@ export class ForecastRepository {
const previous = await tx.tbankForecast.findUnique({ const previous = await tx.tbankForecast.findUnique({
where: { userId_revision_scopeKey: key }, 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 = const fallback =
previous?.membershipRevision === input.membershipRevision saved?.membershipRevision === input.membershipRevision
? previous ? saved
: null; : null;
if (!input.candidate) return fallback; if (!input.candidate) return fallback;
if (!previous && fallback) return fallback;
const portfolios = await tx.tbankPortfolio.findMany({ const portfolios = await tx.tbankPortfolio.findMany({
where: { where: {
...owner, ...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;
}
}
@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { TbankModule } from '../../../integrations/tbank/tbank.module';
import { AuthModule } from '../../auth/auth.module'; import { AuthModule } from '../../auth/auth.module';
import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module';
import { ForecastService } from './application/forecast.service'; import { ForecastService } from './application/forecast.service';
import { PortfolioService } from './application/portfolio.service'; import { PortfolioService } from './application/portfolio.service';
import { ForecastRepository } from './infrastructure/forecast.repository'; import { ForecastRepository } from './infrastructure/forecast.repository';
@@ -9,7 +10,7 @@ import { PortfolioRepository } from './infrastructure/portfolio.repository';
import { PortfolioController } from './presentation/portfolio.controller'; import { PortfolioController } from './presentation/portfolio.controller';
@Module({ @Module({
imports: [AuthModule, TbankModule], imports: [AuthModule, TbankModule, TbankSyncRepositoryModule],
controllers: [PortfolioController], controllers: [PortfolioController],
providers: [ providers: [
PortfolioService, PortfolioService,
@@ -0,0 +1,9 @@
import { Module } from '@nestjs/common';
import { TbankSyncRepository } from './tbank-sync.repository';
@Module({
providers: [TbankSyncRepository],
exports: [TbankSyncRepository],
})
export class TbankSyncRepositoryModule {}
@@ -22,6 +22,7 @@ import {
} from 'class-validator'; } from 'class-validator';
import { MAX_INSTRUMENT_QUERY_LENGTH, MIN_INSTRUMENT_QUERY_LENGTH } from '../instrument-query'; import { MAX_INSTRUMENT_QUERY_LENGTH, MIN_INSTRUMENT_QUERY_LENGTH } from '../instrument-query';
import { InvestmentAttemptProblemDto } from '../investment-resource.dto';
import { operationCategories } from '../operations/operations.api'; import { operationCategories } from '../operations/operations.api';
const MIN_ACCOUNT_ID_LENGTH = 1; const MIN_ACCOUNT_ID_LENGTH = 1;
@@ -267,6 +268,6 @@ export class HourlySyncStatusDto {
@ApiProperty({ type: String, nullable: true, format: 'date-time' }) @ApiProperty({ type: String, nullable: true, format: 'date-time' })
declare finishedAt: null | string; declare finishedAt: null | string;
@ApiProperty({ type: HourlySyncSectionDto, isArray: true }) @ApiProperty({ type: InvestmentAttemptProblemDto, isArray: true })
declare sections: HourlySyncSectionDto[]; declare problems: InvestmentAttemptProblemDto[];
} }
@@ -7,8 +7,8 @@ import { OperationsModule } from '../operations/operations.module';
import { PortfolioModule } from '../portfolio/portfolio.module'; import { PortfolioModule } from '../portfolio/portfolio.module';
import { UpcomingEventsModule } from '../upcoming-events/upcoming-events.module'; import { UpcomingEventsModule } from '../upcoming-events/upcoming-events.module';
import { TbankHourlyScheduler } from './tbank-hourly.scheduler'; import { TbankHourlyScheduler } from './tbank-hourly.scheduler';
import { TbankSyncRepositoryModule } from './tbank-sync-repository.module';
import { TbankSyncController } from './tbank-sync.controller'; import { TbankSyncController } from './tbank-sync.controller';
import { TbankSyncRepository } from './tbank-sync.repository';
import { TbankSyncService } from './tbank-sync.service'; import { TbankSyncService } from './tbank-sync.service';
@Module({ @Module({
imports: [ imports: [
@@ -18,9 +18,10 @@ import { TbankSyncService } from './tbank-sync.service';
OperationsModule, OperationsModule,
InstrumentSourcesModule, InstrumentSourcesModule,
UpcomingEventsModule, UpcomingEventsModule,
TbankSyncRepositoryModule,
], ],
controllers: [TbankSyncController], controllers: [TbankSyncController],
providers: [TbankSyncService, TbankSyncRepository, TbankHourlyScheduler], providers: [TbankSyncService, TbankHourlyScheduler],
exports: [TbankHourlyScheduler, TbankSyncRepository], exports: [TbankHourlyScheduler],
}) })
export class TbankSyncModule {} export class TbankSyncModule {}
@@ -3,6 +3,7 @@ import { randomUUID } from 'node:crypto';
import type { Prisma } from '../../../../generated/prisma/client'; import type { Prisma } from '../../../../generated/prisma/client';
import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api'; import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api';
import type { InvestmentAttemptDto } from '../investment-resource.dto';
import type { InvestmentScreenDto } from './tbank-sync.dto'; import type { InvestmentScreenDto } from './tbank-sync.dto';
import { PrismaService } from '../../../prisma/prisma.service'; import { PrismaService } from '../../../prisma/prisma.service';
@@ -21,6 +22,40 @@ export type HourlyProgress = {
result?: 'failed' | 'partial' | 'succeeded'; result?: 'failed' | 'partial' | 'succeeded';
sections: HourlySection[]; 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() @Injectable()
export class TbankSyncRepository { export class TbankSyncRepository {
constructor(private readonly prisma: PrismaService) {} constructor(private readonly prisma: PrismaService) {}
@@ -4,11 +4,12 @@ import { TbankModule } from '../../../integrations/tbank/tbank.module';
import { AuthModule } from '../../auth/auth.module'; import { AuthModule } from '../../auth/auth.module';
import { InstrumentSourcesModule } from '../instruments/instrument-sources.module'; import { InstrumentSourcesModule } from '../instruments/instrument-sources.module';
import { PortfolioModule } from '../portfolio/portfolio.module'; import { PortfolioModule } from '../portfolio/portfolio.module';
import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module';
import { UpcomingEventsController } from './upcoming-events.controller'; import { UpcomingEventsController } from './upcoming-events.controller';
import { UpcomingEventsService } from './upcoming-events.service'; import { UpcomingEventsService } from './upcoming-events.service';
@Module({ @Module({
imports: [AuthModule, TbankModule, InstrumentSourcesModule, PortfolioModule], imports: [AuthModule, TbankModule, InstrumentSourcesModule, PortfolioModule, TbankSyncRepositoryModule],
controllers: [UpcomingEventsController], controllers: [UpcomingEventsController],
providers: [UpcomingEventsService], providers: [UpcomingEventsService],
exports: [UpcomingEventsService], exports: [UpcomingEventsService],
@@ -20,6 +20,10 @@ import {
type PortfolioPositionDto, type PortfolioPositionDto,
PortfolioService, PortfolioService,
} from '../portfolio/portfolio.api'; } from '../portfolio/portfolio.api';
import {
hourlyAttempt,
TbankSyncRepository,
} from '../tbank-sync/tbank-sync.repository';
import { bondEvent, dividendEvent } from './upcoming-event.mapper'; import { bondEvent, dividendEvent } from './upcoming-event.mapper';
import { import {
UpcomingEventDto, UpcomingEventDto,
@@ -43,6 +47,7 @@ export class UpcomingEventsService {
private readonly sources: CatalogEnrichmentService, private readonly sources: CatalogEnrichmentService,
private readonly forecasts: ForecastService, private readonly forecasts: ForecastService,
private readonly metadata: InstrumentMetadataService, private readonly metadata: InstrumentMetadataService,
private readonly sync: TbankSyncRepository,
) {} ) {}
// Read endpoints stay local; callers without a saved portfolio receive unknown. // Read endpoints stay local; callers without a saved portfolio receive unknown.
@@ -54,6 +59,7 @@ export class UpcomingEventsService {
const window = dateWindow(query); const window = dateWindow(query);
const connection = await this.access.local(userId, accountId); const connection = await this.access.local(userId, accountId);
const portfolio = await this.portfolios.read(userId, accountId); const portfolio = await this.portfolios.read(userId, accountId);
const hourly = await this.sync.hourly(userId);
const events: UpcomingEventDto[] = []; const events: UpcomingEventDto[] = [];
let ready = portfolio.data !== null; let ready = portfolio.data !== null;
const positions = portfolio.data?.positions.filter(supported) ?? []; const positions = portfolio.data?.positions.filter(supported) ?? [];
@@ -61,7 +67,7 @@ export class UpcomingEventsService {
userId, userId,
positions, positions,
window, window,
query.sync === 'observe', true,
); );
for (const position of positions) { for (const position of positions) {
const generic = metadata.get(metadataKey(position))?.data; const generic = metadata.get(metadataKey(position))?.data;
@@ -98,7 +104,7 @@ export class UpcomingEventsService {
source.input.uid === uid && source.input.kind === 'detail', source.input.uid === uid && source.input.kind === 'detail',
)?.instrument || generic; )?.instrument || generic;
const ticker = position.ticker || instrument?.ticker; const ticker = position.ticker || instrument?.ticker;
if (!calendar || calendar.meta.availability !== 'complete' || !ticker) { if (!calendar || calendar.meta.completeness !== 'complete' || !ticker) {
ready = false; ready = false;
continue; continue;
} }
@@ -157,6 +163,11 @@ export class UpcomingEventsService {
membershipRevision: connection.membershipRevision, membershipRevision: connection.membershipRevision,
scopeKey: `events/${JSON.stringify([accountId, window.from, window.to])}`, scopeKey: `events/${JSON.stringify([accountId, window.from, window.to])}`,
portfolio: portfolio.meta, portfolio: portfolio.meta,
lastAttempt: hourlyAttempt(
hourly,
'events',
accountId === 'all' ? undefined : accountId,
),
sources, sources,
candidate, candidate,
}); });
@@ -168,8 +179,12 @@ export class UpcomingEventsService {
// Triggers online refreshes of the account's event sources and returns the // Triggers online refreshes of the account's event sources and returns the
// launched work ids for the hourly walk to observe. // launched work ids for the hourly walk to observe.
async refresh(userId: string, accountId: string): Promise<string[]> { async refresh(
const window = dateWindow({}); userId: string,
accountId: string,
query: UpcomingEventsQueryDto = {},
): Promise<string[]> {
const window = dateWindow(query);
await this.access.local(userId, accountId); await this.access.local(userId, accountId);
const portfolio = await this.portfolios.read(userId, accountId); const portfolio = await this.portfolios.read(userId, accountId);
const positions = portfolio.data?.positions.filter(supported) ?? []; const positions = portfolio.data?.positions.filter(supported) ?? [];
+1 -1
View File
@@ -81,7 +81,7 @@ describe('All accounts HTTP boundary', () => {
expect(response.body).not.toHaveProperty('token'); expect(response.body).not.toHaveProperty('token');
expect(response.body).not.toHaveProperty('generation'); expect(response.body).not.toHaveProperty('generation');
expect(response.body).toMatchObject({ expect(response.body).toMatchObject({
meta: { availability: 'complete' }, meta: { availability: 'available', completeness: 'complete' },
...(path === 'portfolio' ? { data: { positions: [] } } : {}), ...(path === 'portfolio' ? { data: { positions: [] } } : {}),
}); });
}, },
+3 -2
View File
@@ -240,7 +240,8 @@ describe('atomic shared candle reconciliation (SQLite)', () => {
data: { fetchedAt: new Date(0) }, data: { fetchedAt: new Date(0) },
}); });
const stale = await reopened.read('owner', input, true); const stale = await reopened.read('owner', input, true);
expect(stale.meta.availability).toBe('complete'); expect(stale.meta.availability).toBe('available');
expect(stale.meta.completeness).toBe('complete');
expect(stale.meta.freshness).toBe('stale'); expect(stale.meta.freshness).toBe('stale');
expect(stale.meta.coverage?.missingRanges).toEqual([]); expect(stale.meta.coverage?.missingRanges).toEqual([]);
expect(stale.meta.coverage?.completeRanges.length).toBeGreaterThan(0); expect(stale.meta.coverage?.completeRanges.length).toBeGreaterThan(0);
@@ -302,7 +303,7 @@ describe('atomic shared candle reconciliation (SQLite)', () => {
expect(await prisma.tbankCandleCoverage.findMany()).toEqual(before); expect(await prisma.tbankCandleCoverage.findMany()).toEqual(before);
expect(await prisma.tbankCandle.count()).toBe(0); expect(await prisma.tbankCandle.count()).toBe(0);
const read = await service.read('owner', input, true); const read = await service.read('owner', input, true);
expect(read.meta.sync.state).toBe('failed'); expect(read.meta.lastAttempt.state).toBe('failed');
expect(read.data?.candles).toEqual([]); expect(read.data?.candles).toEqual([]);
}); });
it.each([ it.each([
+3 -2
View File
@@ -85,14 +85,15 @@ describe('shared catalog discovery (SQLite)', () => {
it('persists normalized query identity and confirmed empty separately from unknown, across users and reopen', async () => { it('persists normalized query identity and confirmed empty separately from unknown, across users and reopen', async () => {
const cold = await service.read('owner', { ...query, query: ' sber ' }); const cold = await service.read('owner', { ...query, query: ' sber ' });
expect(cold.data).toBeNull(); expect(cold.data).toBeNull();
expect(cold.meta.availability).toBe('missing'); expect(cold.meta.availability).toBe('initial-loading');
await settled(); await settled();
await prisma.$disconnect(); await prisma.$disconnect();
service = createService(); service = createService();
const decrypt = jest.spyOn(tokens, 'decrypt'); const decrypt = jest.spyOn(tokens, 'decrypt');
const warm = await service.read('other', query); const warm = await service.read('other', query);
expect(warm.data).toEqual([instrument]); expect(warm.data).toEqual([instrument]);
expect(warm.meta.availability).toBe('complete'); expect(warm.meta.availability).toBe('available');
expect(warm.meta.completeness).toBe('complete');
expect(warm.meta.sync.id).toBeNull(); expect(warm.meta.sync.id).toBeNull();
expect(decrypt).not.toHaveBeenCalled(); expect(decrypt).not.toHaveBeenCalled();
expect(upstream.findShares).toHaveBeenCalledTimes(1); expect(upstream.findShares).toHaveBeenCalledTimes(1);
+11 -8
View File
@@ -128,7 +128,8 @@ describe('local income completeness and exact money (SQLite)', () => {
for (let index = 0; index < 2; index++) { for (let index = 0; index < 2; index++) {
const result = await read(); const result = await read();
assert(result.data); assert(result.data);
expect(result.meta.availability).toBe('complete'); expect(result.meta.availability).toBe('available');
expect(result.meta.completeness).toBe('complete');
expect(result.data.totalsByCurrency.map((row) => row.total)).toEqual([ expect(result.data.totalsByCurrency.map((row) => row.total)).toEqual([
{ currency: 'RUB', value: '15.000000002' }, { currency: 'RUB', value: '15.000000002' },
{ currency: 'USD', value: '6' }, { currency: 'USD', value: '6' },
@@ -146,10 +147,9 @@ describe('local income completeness and exact money (SQLite)', () => {
'previous', 'previous',
'year', 'year',
]); ]);
expect( const previous = result.meta.dependencies.find(({ role }) => role === 'previous');
result.meta.dependencies.find(({ role }) => role === 'previous') expect(previous?.availability).toBe('temporarily-unavailable');
?.availability, expect(previous?.completeness).toBe('incomplete');
).toBe('missing');
expect(result.data.calculatedAt).toBe('2025-02-01T00:00:00.000Z'); expect(result.data.calculatedAt).toBe('2025-02-01T00:00:00.000Z');
await prisma.$disconnect(); await prisma.$disconnect();
} }
@@ -161,7 +161,9 @@ describe('local income completeness and exact money (SQLite)', () => {
where: { accountId: 'b' }, where: { accountId: 'b' },
}); });
expect((await read()).data).toBeNull(); expect((await read()).data).toBeNull();
expect((await read()).meta.availability).toBe('partial'); const incomplete = await read();
expect(incomplete.meta.availability).toBe('temporarily-unavailable');
expect(incomplete.meta.completeness).toBe('incomplete');
expect((await read('a')).data).not.toBeNull(); expect((await read('a')).data).not.toBeNull();
expect((await read('a', 'from=2025-03-01&to=2025-03-31')).data).toBeNull(); expect((await read('a', 'from=2025-03-01&to=2025-03-31')).data).toBeNull();
await coverage( await coverage(
@@ -241,7 +243,7 @@ describe('local income completeness and exact money (SQLite)', () => {
await repo.fail(fence, false); await repo.fail(fence, false);
const after = await read(); const after = await read();
expect(after.data).toEqual(before.data); expect(after.data).toEqual(before.data);
expect(after.meta.sync.state).toBe('failed'); expect(after.meta.lastAttempt.state).toBe('failed');
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled(); expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
}); });
it('retains current-month observed completeness and does not renew source timestamps', async () => { it('retains current-month observed completeness and does not renew source timestamps', async () => {
@@ -261,7 +263,8 @@ describe('local income completeness and exact money (SQLite)', () => {
from: `${today.slice(0, 7)}-01`, from: `${today.slice(0, 7)}-01`,
to: today, to: today,
}); });
expect(result.meta.availability).toBe('complete'); expect(result.meta.availability).toBe('available');
expect(result.meta.completeness).toBe('complete');
expect(result.meta.coverage?.asOf).toBe(asOf.toISOString()); expect(result.meta.coverage?.asOf).toBe(asOf.toISOString());
expect(result.data?.calculatedAt).toBe('2025-02-01T00:00:00.000Z'); expect(result.data?.calculatedAt).toBe('2025-02-01T00:00:00.000Z');
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled(); expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
+17 -9
View File
@@ -118,7 +118,8 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
`limit=2${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`, `limit=2${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`,
).expect(200); ).expect(200);
const result = response.body as BrokerageOperationsResourceDto; const result = response.body as BrokerageOperationsResourceDto;
expect(result.meta.availability).toBe('complete'); expect(result.meta.availability).toBe('available');
expect(result.meta.completeness).toBe('complete');
assert(result.data); assert(result.data);
ids.push( ids.push(
...result.data.items.map((item) => `${item.account?.id}/${item.id}`), ...result.data.items.map((item) => `${item.account?.id}/${item.id}`),
@@ -251,15 +252,18 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
.set('Authorization', auth()); .set('Authorization', auth());
const first = (await page().expect(200)) const first = (await page().expect(200))
.body as BrokerageOperationsResourceDto; .body as BrokerageOperationsResourceDto;
expect(first.meta.availability).toBe('complete'); expect(first.meta.availability).toBe('available');
expect(first.meta.completeness).toBe('complete');
assert(first.data?.nextCursor); assert(first.data?.nextCursor);
const second = (await page(first.data.nextCursor).expect(200)) const second = (await page(first.data.nextCursor).expect(200))
.body as BrokerageOperationsResourceDto; .body as BrokerageOperationsResourceDto;
expect(second.meta.availability).toBe('complete'); expect(second.meta.availability).toBe('available');
expect(second.meta.completeness).toBe('complete');
expect(second.meta.coverage?.missingRanges).toEqual([]); expect(second.meta.coverage?.missingRanges).toEqual([]);
const observed = (await page(first.data.nextCursor, true).expect(200)) const observed = (await page(first.data.nextCursor, true).expect(200))
.body as BrokerageOperationsResourceDto; .body as BrokerageOperationsResourceDto;
expect(observed.meta.availability).toBe('complete'); expect(observed.meta.availability).toBe('available');
expect(observed.meta.completeness).toBe('complete');
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled(); expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
}, },
); );
@@ -384,12 +388,14 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
'a', 'a',
).expect(200) ).expect(200)
).body as BrokerageOperationsResourceDto; ).body as BrokerageOperationsResourceDto;
expect(partial.meta.availability).toBe('partial'); expect(partial.meta.availability).toBe('initial-loading');
expect(partial.data?.hasNext).toBe(false); expect(partial.meta.completeness).toBe('incomplete');
expect(partial.data).toBeNull();
expect(partial.meta.coverage?.missingRanges.length).toBeGreaterThan(0); expect(partial.meta.coverage?.missingRanges.length).toBeGreaterThan(0);
const whole = (await get('sync=observe', 'a').expect(200)) const whole = (await get('sync=observe', 'a').expect(200))
.body as BrokerageOperationsResourceDto; .body as BrokerageOperationsResourceDto;
expect(whole.meta.availability).toBe('complete'); expect(whole.meta.availability).toBe('available');
expect(whole.meta.completeness).toBe('complete');
expect(whole.data?.items).toEqual([]); expect(whole.data?.items).toEqual([]);
expect(upstream.getOperationsByCursor.mock.calls.length).toBeGreaterThan(0); expect(upstream.getOperationsByCursor.mock.calls.length).toBeGreaterThan(0);
}); });
@@ -419,7 +425,8 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
const cold = (await get('', 'a').expect(200)) const cold = (await get('', 'a').expect(200))
.body as BrokerageOperationsResourceDto; .body as BrokerageOperationsResourceDto;
expect(cold.data).toBeNull(); expect(cold.data).toBeNull();
expect(cold.meta.availability).toBe('missing'); expect(cold.meta.availability).toBe('initial-loading');
expect(cold.meta.completeness).toBe('incomplete');
await get('sync=observe', 'a').expect(200); await get('sync=observe', 'a').expect(200);
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1); expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
// eslint-disable-next-line sonarjs/no-extra-arguments -- The typed callback is assigned the promise resolver before invocation. // eslint-disable-next-line sonarjs/no-extra-arguments -- The typed callback is assigned the promise resolver before invocation.
@@ -434,7 +441,8 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
const result = (await get('sync=observe', 'a').expect(200)) const result = (await get('sync=observe', 'a').expect(200))
.body as BrokerageOperationsResourceDto; .body as BrokerageOperationsResourceDto;
expect(result.data?.items).toEqual([]); expect(result.data?.items).toEqual([]);
expect(result.meta.availability).toBe('complete'); expect(result.meta.availability).toBe('available');
expect(result.meta.completeness).toBe('complete');
await get('sync=observe', 'a').expect(200); await get('sync=observe', 'a').expect(200);
expect(upstream.getOperationsByCursor.mock.calls.length).toBeGreaterThanOrEqual(count); expect(upstream.getOperationsByCursor.mock.calls.length).toBeGreaterThanOrEqual(count);
}); });
+9 -8
View File
@@ -151,7 +151,7 @@ describe('local accounts and portfolio HTTP (SQLite)', () => {
const cold = await get('brokerage-accounts/unknown/portfolio?sync=observe').expect(200); const cold = await get('brokerage-accounts/unknown/portfolio?sync=observe').expect(200);
expect((cold.body as PortfolioResourceDto).data).toBeNull(); expect((cold.body as PortfolioResourceDto).data).toBeNull();
expect((cold.body as PortfolioResourceDto).meta.availability).toBe( expect((cold.body as PortfolioResourceDto).meta.availability).toBe(
'missing', 'temporarily-unavailable',
); );
await get('brokerage-accounts?sync=observe').expect(200); await get('brokerage-accounts?sync=observe').expect(200);
expect(upstream.getOpenAccounts).not.toHaveBeenCalled(); expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
@@ -321,7 +321,7 @@ describe('local accounts and portfolio HTTP (SQLite)', () => {
'brokerage-accounts/account/portfolio?sync=observe', 'brokerage-accounts/account/portfolio?sync=observe',
).expect(200); ).expect(200);
expect((failed.body as PortfolioResourceDto).data).toEqual(previous); expect((failed.body as PortfolioResourceDto).data).toEqual(previous);
expect((failed.body as PortfolioResourceDto).meta.sync.state).toBe('idle'); expect((failed.body as PortfolioResourceDto).meta.lastAttempt.state).toBe('none');
expect(upstream.getPortfolio).toHaveBeenCalledTimes(1); expect(upstream.getPortfolio).toHaveBeenCalledTimes(1);
}); });
@@ -375,7 +375,7 @@ describe('local accounts and portfolio HTTP (SQLite)', () => {
where: { id: 'abandoned-membership' }, where: { id: 'abandoned-membership' },
data: { state: 'complete' }, data: { state: 'complete' },
}); });
expect((await get('brokerage-accounts/account/portfolio?sync=observe').expect(200)).body).toMatchObject({ data: null, meta: { availability: 'missing' } }); expect((await get('brokerage-accounts/account/portfolio?sync=observe').expect(200)).body).toMatchObject({ data: null, meta: { availability: 'temporarily-unavailable', completeness: 'incomplete' } });
expect(await prisma.tbankPortfolio.count()).toBe(0); expect(await prisma.tbankPortfolio.count()).toBe(0);
expect((await get('brokerage-accounts?sync=observe')).body).toMatchObject({ expect((await get('brokerage-accounts?sync=observe')).body).toMatchObject({
data: [{ id: 'account', portfolioSummary: null }], data: [{ id: 'account', portfolioSummary: null }],
@@ -423,13 +423,14 @@ describe('local accounts and portfolio HTTP (SQLite)', () => {
{ currency: 'RUB', value: '21.1' }, { currency: 'RUB', value: '21.1' },
]); ]);
expect(summaries.data?.[1].portfolioSummary).toBeNull(); expect(summaries.data?.[1].portfolioSummary).toBeNull();
expect(summaries.meta.availability).toBe('partial'); expect(summaries.meta.availability).toBe('available');
expect(summaries.meta.completeness).toBe('complete');
expect( expect(
(await get('brokerage-accounts/all/portfolio?sync=observe')).body, (await get('brokerage-accounts/all/portfolio?sync=observe')).body,
).toMatchObject({ data: null, meta: { availability: 'partial' } }); ).toMatchObject({ data: null, meta: { availability: 'temporarily-unavailable', completeness: 'incomplete' } });
expect( expect(
(await get(`stocks/${uid}/position?sync=observe`).expect(200)).body, (await get(`stocks/${uid}/position?sync=observe`).expect(200)).body,
).toMatchObject({ data: null, meta: { availability: 'partial' } }); ).toMatchObject({ data: null, meta: { availability: 'temporarily-unavailable', completeness: 'incomplete' } });
expect(upstream.getPortfolio).toHaveBeenCalledTimes(2); expect(upstream.getPortfolio).toHaveBeenCalledTimes(2);
await prisma.tbankAccount.deleteMany({ where: { accountId: 'second' } }); await prisma.tbankAccount.deleteMany({ where: { accountId: 'second' } });
await prisma.tbankSyncWork.updateMany({ await prisma.tbankSyncWork.updateMany({
@@ -444,7 +445,7 @@ describe('local accounts and portfolio HTTP (SQLite)', () => {
); );
expect(position.body).toMatchObject({ expect(position.body).toMatchObject({
data: { quantity: '2', accounts: [{ account: { id: 'account' } }] }, data: { quantity: '2', accounts: [{ account: { id: 'account' } }] },
meta: { availability: 'complete' }, meta: { availability: 'available', completeness: 'complete' },
}); });
await prisma.$disconnect(); await prisma.$disconnect();
expect( expect(
@@ -495,7 +496,7 @@ describe('local accounts and portfolio HTTP (SQLite)', () => {
const empty = await get('brokerage-accounts?sync=observe').expect(200); const empty = await get('brokerage-accounts?sync=observe').expect(200);
expect((empty.body as BrokerageAccountsResourceDto).data).toEqual([]); expect((empty.body as BrokerageAccountsResourceDto).data).toEqual([]);
expect((empty.body as BrokerageAccountsResourceDto).meta.availability).toBe( expect((empty.body as BrokerageAccountsResourceDto).meta.availability).toBe(
'complete', 'available',
); );
await put().expect(200); await put().expect(200);
await seedMembership(); await seedMembership();
+56 -38
View File
@@ -21,6 +21,7 @@ import { configureApp } from '../src/main';
import { CatalogEnrichmentService } from '../src/modules/investments/instruments/application/catalog-enrichment.service'; import { CatalogEnrichmentService } from '../src/modules/investments/instruments/application/catalog-enrichment.service';
import { ForecastRepository } from '../src/modules/investments/portfolio/infrastructure/forecast.repository'; import { ForecastRepository } from '../src/modules/investments/portfolio/infrastructure/forecast.repository';
import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler'; import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler';
import { UpcomingEventsService } from '../src/modules/investments/upcoming-events/upcoming-events.service';
import { PrismaService } from '../src/prisma/prisma.service'; import { PrismaService } from '../src/prisma/prisma.service';
const shareUid = '00000000-0000-4000-8000-000000000001'; const shareUid = '00000000-0000-4000-8000-000000000001';
@@ -81,6 +82,10 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
`brokerage-accounts/${scope}/events?from=${range.from}&to=${range.to}${observe ? '&sync=observe' : ''}`, `brokerage-accounts/${scope}/events?from=${range.from}&to=${range.to}${observe ? '&sync=observe' : ''}`,
userId, userId,
); );
const refresh = async (range = window, accountId = 'account') => {
await app.get(UpcomingEventsService).refresh('owner', accountId, range);
await settled();
};
async function settled() { async function settled() {
for (let i = 0; i < 500; i++) { for (let i = 0; i < 500; i++) {
await delay(5); await delay(5);
@@ -235,39 +240,29 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
}); });
it('returns unknown immediately, keeps observe passive and shares confirmed empty calendars across users', async () => { it('returns unknown immediately, keeps observe passive and shares confirmed empty calendars across users', async () => {
let release: (value: unknown[]) => void = () => {};
upstream.getDividends.mockReturnValueOnce(
new Promise<unknown[]>((resolve) => {
release = resolve;
}),
);
expect( expect(
((await events().expect(200)).body as UpcomingEventsResourceDto).data, ((await events().expect(200)).body as UpcomingEventsResourceDto).data,
).toBeNull(); ).toBeNull();
expect(upstream.getDividends).not.toHaveBeenCalled(); expect(upstream.getDividends).not.toHaveBeenCalled();
const cold = (await events(false).expect(200)) const cold = (await events().expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
expect(cold.data).toBeNull(); expect(cold.data).toBeNull();
for (let i = 0; i < 100 && !upstream.getDividends.mock.calls.length; i++) { expect(await prisma.tbankForecast.count()).toBe(0);
await delay(5); });
}
it('keeps an ordinary GET local and does not start catalog work', async () => {
await portfolio([position()]);
jest.clearAllMocks();
await events().expect(200); await events().expect(200);
release([]);
await settled(); expect(upstream.getDividends).not.toHaveBeenCalled();
const warm = (await events().expect(200)).body as UpcomingEventsResourceDto; expect(upstream.getInstrumentByUid).not.toHaveBeenCalled();
expect(warm.meta.availability).toBe('complete');
expect(warm.data?.events).toEqual([]);
const other = (await events(false, 'account', 'other').expect(200))
.body as UpcomingEventsResourceDto;
expect(other.data?.events).toEqual([]);
expect(upstream.getDividends).toHaveBeenCalledTimes(1);
expect(await prisma.tbankForecast.count()).toBe(2);
expect(JSON.stringify(warm)).not.toContain('owner-revision');
}); });
it('preserves source identity, inclusive dates, exact currencies, cancellation and offers in account/all scopes', async () => { it('preserves source identity, inclusive dates, exact currencies, cancellation and offers in account/all scopes', async () => {
await portfolio([position(), position(bondUid, '3', 'BOND')]); await portfolio([position(), position(bondUid, '3', 'BOND')]);
await events(false).expect(200); await app.get(UpcomingEventsService).refresh('owner', 'account', window);
await settled(); await settled();
const single = (await events().expect(200)) const single = (await events().expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
@@ -297,7 +292,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
}); });
it('keeps the previous full forecast and timestamps after a new holding calendar fails, including reopen', async () => { it('keeps the previous full forecast and timestamps after a new holding calendar fails, including reopen', async () => {
await events(false).expect(200); await refresh();
await settled(); await settled();
const before = (await events().expect(200)) const before = (await events().expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
@@ -307,7 +302,8 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
? Promise.reject(new Error('new calendar unavailable')) ? Promise.reject(new Error('new calendar unavailable'))
: Promise.resolve([]), : Promise.resolve([]),
); );
const pending = (await events(false).expect(200)) await refresh();
const pending = (await events().expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
expect(pending.data).toEqual(before.data); expect(pending.data).toEqual(before.data);
await settled(); await settled();
@@ -316,14 +312,14 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
expect(after.data).toEqual(before.data); expect(after.data).toEqual(before.data);
expect(after.meta.lastSuccessfulAt).toBe(before.meta.lastSuccessfulAt); expect(after.meta.lastSuccessfulAt).toBe(before.meta.lastSuccessfulAt);
expect(after.meta.sync.state).toBe('failed'); expect(after.meta.dependencies?.some((dependency) => dependency.lastAttempt.state === 'failed')).toBe(true);
const count = upstream.getDividends.mock.calls.length; const count = upstream.getDividends.mock.calls.length;
await events().expect(200); await events().expect(200);
expect(upstream.getDividends).toHaveBeenCalledTimes(count); expect(upstream.getDividends).toHaveBeenCalledTimes(count);
}); });
it('never relabels another range or membership snapshot and rejects inaccessible accounts locally', async () => { it('never relabels another range or membership snapshot and rejects inaccessible accounts locally', async () => {
await events(false).expect(200); await refresh();
await settled(); await settled();
await events().expect(200); await events().expect(200);
const newRange = ( const newRange = (
@@ -346,7 +342,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
}); });
it('refuses a calculated snapshot if its portfolio changes before atomic publication', async () => { it('refuses a calculated snapshot if its portfolio changes before atomic publication', async () => {
await events(false).expect(200); await refresh();
await settled(); await settled();
const before = (await events().expect(200)) const before = (await events().expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
@@ -531,7 +527,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
it('keeps optional metadata failure separate from calendar completeness and uses portfolio ticker', async () => { it('keeps optional metadata failure separate from calendar completeness and uses portfolio ticker', async () => {
upstream.getBond.mockRejectedValue(new Error('metadata unavailable')); upstream.getBond.mockRejectedValue(new Error('metadata unavailable'));
await portfolio([position(bondUid, '3', 'BOND')]); await portfolio([position(bondUid, '3', 'BOND')]);
await events(false).expect(200); await refresh();
await settled(); await settled();
const result = (await events().expect(200)) const result = (await events().expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
@@ -566,7 +562,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
new MoexClient().bondEvents(ticker), new MoexClient().bondEvents(ticker),
); );
try { try {
await events(false).expect(200); await refresh();
await settled(); await settled();
const before = (await events().expect(200)) const before = (await events().expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
@@ -590,7 +586,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
expect(after.data).toEqual(before.data); expect(after.data).toEqual(before.data);
expect(after.meta.lastSuccessfulAt).toBe(before.meta.lastSuccessfulAt); expect(after.meta.lastSuccessfulAt).toBe(before.meta.lastSuccessfulAt);
expect(after.meta.sync.state).toBe('failed'); expect(after.meta.dependencies?.some((dependency) => dependency.lastAttempt.state === 'failed')).toBe(true);
expect( expect(
await prisma.tbankInstrumentSource.findFirstOrThrow({ await prisma.tbankInstrumentSource.findFirstOrThrow({
where: { kind: 'bond-events' }, where: { kind: 'bond-events' },
@@ -605,9 +601,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
it('defaults to a 30-day window', async () => { it('defaults to a 30-day window', async () => {
await portfolio([position(bondUid, '3', 'BOND')]); await portfolio([position(bondUid, '3', 'BOND')]);
await events(false).expect(200); await refresh();
await settled();
await get('brokerage-accounts/account/events?from=2099-02-01').expect(200);
await settled(); await settled();
const result = ( const result = (
await get( await get(
@@ -622,6 +616,30 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
]); ]);
}); });
it('filters a saved source window without changing its calculated timestamp', async () => {
await portfolio([position(bondUid, '3', 'BOND')]);
await app.get(UpcomingEventsService).refresh('owner', 'account', {
from: '2099-02-01',
to: '2099-02-28',
});
await settled();
const full = (await events().expect(200)).body as UpcomingEventsResourceDto;
const calculatedAt = full.data?.calculatedAt;
jest.clearAllMocks();
const narrow = (await events(false, 'account', 'owner', {
from: '2099-02-01',
to: '2099-02-10',
}).expect(200)).body as UpcomingEventsResourceDto;
expect(narrow.data?.calculatedAt).toBe(calculatedAt);
expect(narrow.data?.events.map((event) => event.date)).toEqual([
'2099-02-01',
]);
expect(upstream.getDividends).not.toHaveBeenCalled();
expect(moex.bondEvents).not.toHaveBeenCalled();
});
it('discards a late calendar from a replaced connection and prevents access to old private forecasts', async () => { it('discards a late calendar from a replaced connection and prevents access to old private forecasts', async () => {
let release: (value: unknown[]) => void = () => {}; let release: (value: unknown[]) => void = () => {};
upstream.getDividends.mockReturnValueOnce( upstream.getDividends.mockReturnValueOnce(
@@ -629,7 +647,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
release = resolve; release = resolve;
}), }),
); );
await events(false).expect(200); await app.get(UpcomingEventsService).refresh('owner', 'account', window);
for (let i = 0; i < 100 && !upstream.getDividends.mock.calls.length; i++) { for (let i = 0; i < 100 && !upstream.getDividends.mock.calls.length; i++) {
await delay(5); await delay(5);
} }
@@ -664,8 +682,7 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
it('uses UTC today for an end-only range and includes the entire leap day', async () => { it('uses UTC today for an end-only range and includes the entire leap day', async () => {
const today = new Date().toISOString().slice(0, 10); const today = new Date().toISOString().slice(0, 10);
await get('brokerage-accounts/account/events?to=2096-02-29').expect(200); await refresh({ from: today, to: '2096-02-29' });
await settled();
expect(upstream.getDividends).toHaveBeenCalledWith( expect(upstream.getDividends).toHaveBeenCalledWith(
'synthetic-owner', 'synthetic-owner',
shareUid, shareUid,
@@ -685,8 +702,8 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
}, },
]); ]);
const range = { from: '2096-02-29', to: '2096-02-29' }; const range = { from: '2096-02-29', to: '2096-02-29' };
await events(false, 'account', 'owner', range).expect(200); await refresh(range);
await settled(); jest.clearAllMocks();
const result = (await events(true, 'account', 'owner', range).expect(200)) const result = (await events(true, 'account', 'owner', range).expect(200))
.body as UpcomingEventsResourceDto; .body as UpcomingEventsResourceDto;
expect(result.data?.events).toHaveLength(1); expect(result.data?.events).toHaveLength(1);
@@ -694,5 +711,6 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
date: '2096-02-29', date: '2096-02-29',
projectedAmount: { currency: 'RUB', value: '14' }, projectedAmount: { currency: 'RUB', value: '14' },
}); });
expect(upstream.getDividends).not.toHaveBeenCalled();
}); });
}); });