WealthLedger/backend
5
0
Fork 0
Code Issues 7 Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

fix: scope income and ledger sync to requested intervals (WealthLedger/backend#57) (#147)
Backend verify / commits (push) Skipped
Details
Backend verify / security (push) Successful in 17s
Details
Backend verify / verify (push) Failing after 5m52s
Details

Browse Source
This commit was merged in pull request #147.
This commit is contained in:
codex-bot
2026-10-28 21:29:53 +03:00
parent e350146749
commit 5e84e1e5f1
7 changed files with 114 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
eslint-suppressions.json
+1 -4
View File
@@ -92,11 +92,8 @@
}
},
"src/modules/investments/investment-sync-plan.ts": {
"sonarjs/cognitive-complexity": {
"count": 1
},
"sonarjs/no-nested-conditional": {
"count": 4
"count": 3
}
},
"src/modules/investments/operations/application/brokerage-operations.service.ts": {
src/modules/investments/investment-sync-plan.spec.ts
+8 -188
View File
@@ -1,196 +1,16 @@
import {
accountHistory,
ledgerRange,
planInvestmentScreen,
} from './investment-sync-plan';
import { accountHistory, ledgerRange } from './investment-sync-plan';
describe('screen dependency and history planning', () => {
// Missing opening dates must not silently narrow the reported history boundary.
describe('account history boundaries', () => {
const now = new Date('2026-09-25T12:00:00Z');
const context = {
membershipKnown: true,
accounts: [{ accountId: 'a', openedOn: '2020-01-01' }],
instruments: [
{
uid: 'uid',
type: 'STOCK' as const,
historyFrom: new Date('2020-01-01Z'),
},
],
};
it('prioritizes discovery, the selected period, then history; Moscow dates are converted once', () => {
const plan = planInvestmentScreen(
{
screen: 'income',
accountId: 'a',
from: '2024-02-01',
to: '2024-02-29',
},
context,
now,
);
expect(plan.sources[0].kind).toBe('membership');
const selected = plan.sources.filter(
(source) => source.kind === 'operations' && source.priority === 'screen',
);
// The single selected range is chunked into month-aligned intervals.
expect(selected).toHaveLength(2);
expect(
selected.every(
(source) =>
source.from != null &&
source.from.toISOString() >= '2024-01-31T21:00:00.000Z' &&
source.to != null &&
source.to.toISOString() <= '2024-02-29T21:00:00.000Z',
),
).toBe(true);
expect(
selected.some(
(source) => source.from?.toISOString() === '2024-01-31T21:00:00.000Z',
),
).toBe(true);
expect(
selected.some(
(source) => source.to?.toISOString() === '2024-02-29T21:00:00.000Z',
),
).toBe(true);
expect(plan.unavailable).toEqual([]);
const firstHistory = plan.sources.findIndex(
(source) => source.priority === 'history',
);
expect(
plan.sources
.slice(firstHistory)
.every((source) => source.priority === 'history'),
).toBe(true);
});
it('force contains only shown dependencies and retains instrument reconciliation predicates', () => {
const plan = planInvestmentScreen(
{
screen: 'stock',
instrumentUid: 'uid',
from: '2026-01-01',
to: '2026-02-01',
candlePeriod: 'year',
},
context,
now,
true,
);
expect(plan.sources.map((source) => source.kind)).toEqual(
expect.arrayContaining([
'membership',
'portfolio',
'operations',
'metadata',
'quote',
'company',
'calendar',
'candles',
]),
);
expect(plan.sources.every((source) => source.priority !== 'history')).toBe(
true,
);
expect(
plan.sources
.filter((source) => source.kind === 'operations')
.every((source) => source.instrumentUid === 'uid'),
).toBe(true);
const accounts = planInvestmentScreen(
{ screen: 'accounts' },
context,
now,
true,
);
expect(
accounts.sources.some((source) => source.kind === 'operations'),
).toBe(false);
});
it('never invents an account or candle history boundary; discovery gates unknown membership', () => {
const unknown = {
...context,
accounts: [{ accountId: 'a', openedOn: null }],
instruments: [],
};
expect(accountHistory(unknown.accounts, now)[0].from.toISOString()).toBe(
const accounts = [{ accountId: 'a', openedOn: null }];
it('retains the unknown opening boundary for history and instrument ledger', () => {
expect(accountHistory(accounts, now)[0].from.toISOString()).toBe(
'1970-01-01T00:00:00.000Z',
);
expect(
ledgerRange(
undefined,
undefined,
'uid',
unknown.accounts,
now,
).from.toISOString(),
ledgerRange(undefined, undefined, 'uid', accounts, now).from.toISOString(),
).toBe('1970-01-01T00:00:00.000Z');
const plan = planInvestmentScreen(
{ screen: 'bond', instrumentUid: 'uid', candlePeriod: 'all' },
unknown,
now,
);
expect(plan.sources.some((source) => source.kind === 'candles')).toBe(
false,
);
expect(plan.unavailable[0].reason).toBe('history_start_unavailable');
expect(
planInvestmentScreen(
{ screen: 'account' },
{ ...context, membershipKnown: false },
now,
).sources,
).toEqual([{ kind: 'membership', priority: 'discovery' }]);
});
it('keeps UTC event days separate from Moscow instrument calendar days at year rollover', () => {
const rollover = new Date('2026-12-31T22:00:00Z');
const events = planInvestmentScreen(
{ screen: 'events' },
context,
rollover,
true,
);
expect(
events.sources.find((source) => source.kind === 'calendar')?.from,
).toEqual(new Date('2026-12-31Z'));
const stock = planInvestmentScreen(
{ screen: 'stock', instrumentUid: 'uid' },
context,
rollover,
true,
);
expect(
stock.sources.find((source) => source.kind === 'calendar'),
).toMatchObject({
from: new Date('2022-01-01Z'),
to: new Date('2028-01-01Z'),
});
expect(rollover.toISOString()).toBe('2026-12-31T22:00:00.000Z');
});
it('keeps search queries separate and events scoped to their inclusive period', () => {
const search = planInvestmentScreen(
{ screen: 'stocks', query: ' sber ' },
context,
now,
true,
);
expect(
search.sources.find((source) => source.kind === 'catalog')?.query,
).toBe('SBER');
const events = planInvestmentScreen(
{ screen: 'events', from: '2026-10-01', to: '2026-10-31' },
context,
now,
true,
);
expect(
events.sources.find((source) => source.kind === 'calendar'),
).toMatchObject({
from: new Date('2026-10-01Z'),
to: new Date('2026-10-31Z'),
});
expect(events.sources.some((source) => source.kind === 'operations')).toBe(
false,
);
});
});
src/modules/investments/investment-sync-plan.ts
+1 -257
View File
@@ -1,267 +1,11 @@
import { ISO_DATE_LENGTH, ISO_DATE_START } from '../../common/time';
import { planIncomePeriod } from './income/income.api';
import { dateOnlyBounds, moscowDate } from './investment-calendar';
import {
type OperationInterval,
operationIntervals,
} from './operations/domain/operations-sync-plan';
import type { OperationInterval } from './operations/domain/operations-sync-plan';
const FORECAST_YEARS = 1;
const JANUARY_INDEX = 0;
const FIRST_DAY_OF_MONTH = 1;
const EXCLUSIVE_END_OFFSET_MS = 1;
const NO_END_OFFSET_MS = 0;
export type InvestmentScreen = {
accountId?: string;
candlePeriod?: 'all' | 'halfYear' | 'month' | 'year';
candleRange?: { from: Date; to: Date };
from?: string;
instrumentUid?: string;
query?: string;
screen:
| 'account' |
'accounts' |
'bond' |
'bonds' |
'events' |
'income' |
'operations' |
'stock' |
'stocks';
to?: string;
};
const defaultEventDays = 30;
const incomeHistoryYears = 5;
const candleMonths = { month: 1, halfYear: 6, year: 12 };
type Account = { accountId: string; openedOn: null | string };
type Instrument = {
historyFrom: Date | null;
type: 'BOND' | 'STOCK';
uid: string;
};
type Source = {
accountId?: string;
from?: Date;
instrumentType?: 'BOND' | 'STOCK';
instrumentUid?: string;
kind:
| 'calendar' |
'candles' |
'catalog' |
'company' |
'membership' |
'metadata' |
'metrics' |
'operations' |
'portfolio' |
'quote';
priority: 'discovery' | 'history' | 'screen';
query?: string;
to?: Date;
};
// Expands screen scope, never credentials or arbitrary caller-supplied jobs.
// Owners execute their own sources and expand instruments after discovery.
export function planInvestmentScreen(
screen: InvestmentScreen,
context: {
accounts: Account[];
instruments: Instrument[];
membershipKnown: boolean;
},
now: Date,
force = false,
) {
const sources: Source[] = [{ kind: 'membership', priority: 'discovery' }];
const accounts = context.accounts.filter(
(account) =>
!screen.accountId ||
screen.accountId === 'all' ||
account.accountId === screen.accountId,
);
const unavailable: { reason: string; resourceKey: string }[] = [];
const operations = (from: Date, to: Date, instrumentUid?: string) => {
const targets = accounts.flatMap((account) =>
from < to
? [
{
accountId: account.accountId,
from,
to,
...(instrumentUid ? { instrumentUid } : {}),
},
]
: [],
);
sources.push(
...operationIntervals(targets).map((range) => ({
...range,
kind: 'operations' as const,
priority: 'screen' as const,
})),
);
};
const portfolio = () =>
sources.push(
...accounts.map(({ accountId }) => ({
kind: 'portfolio' as const,
accountId,
priority: 'screen' as const,
})),
);
if (!context.membershipKnown) return { sources, unavailable, now };
if (
['account', 'accounts', 'bond', 'events', 'stock'].includes(screen.screen)
) {
portfolio();
}
if (
screen.screen === 'operations' ||
screen.screen === 'stock' ||
screen.screen === 'bond'
) {
const range = ledgerRange(
screen.from,
screen.to,
screen.instrumentUid,
accounts,
now,
);
operations(range.from, range.to, force ? screen.instrumentUid : undefined);
}
if (screen.screen === 'account' || screen.screen === 'income') {
const period = planIncomePeriod(
screen.screen === 'income' ? screen.from : undefined,
screen.screen === 'income' ? screen.to : undefined,
);
const bounds = dateOnlyBounds(period.selected, now);
operations(bounds.from, bounds.to);
if (screen.screen === 'account') {
const range = ledgerRange(
screen.from,
screen.to,
undefined,
accounts,
now,
);
operations(range.from, range.to);
}
}
if (screen.screen === 'stocks' || screen.screen === 'bonds') {
sources.push({
kind: 'catalog',
instrumentType: screen.screen === 'stocks' ? 'STOCK' : 'BOND',
query: screen.query?.trim().toUpperCase() ?? '',
priority: 'screen',
});
}
const instruments = context.instruments.filter(
(instrument) =>
!screen.instrumentUid || instrument.uid === screen.instrumentUid,
);
if (
screen.instrumentUid &&
!instruments.length &&
(screen.screen === 'stock' || screen.screen === 'bond')
) {
instruments.push({
uid: screen.instrumentUid,
type: screen.screen === 'stock' ? 'STOCK' : 'BOND',
historyFrom: null,
});
}
for (const instrument of instruments) {
const identity = {
instrumentUid: instrument.uid,
instrumentType: instrument.type,
};
sources.push({ ...identity, kind: 'metadata', priority: 'screen' });
if (['bond', 'bonds', 'stock', 'stocks'].includes(screen.screen)) {
sources.push({ ...identity, kind: 'quote', priority: 'screen' });
}
if (screen.screen === 'stock' || screen.screen === 'bond') {
sources.push({
...identity,
kind: screen.screen === 'stock' ? 'company' : 'metrics',
priority: 'screen',
});
const from = new Date(now);
const period = screen.candlePeriod ?? 'month';
if (period !== 'all') {
// eslint-disable-next-line security/detect-object-injection -- The period belongs to the validated candle-period union.
from.setUTCMonth(from.getUTCMonth() - candleMonths[period]);
}
const range =
screen.candleRange ??
(period === 'all'
? instrument.historyFrom
? { from: instrument.historyFrom, to: now }
: null
: { from, to: now });
if (range) {
sources.push({
...identity,
kind: 'candles',
...range,
priority: 'screen',
});
} else {
unavailable.push({
resourceKey: `candles/${instrument.uid}/all`,
reason: 'history_start_unavailable',
});
}
}
if (['account', 'bond', 'events', 'stock'].includes(screen.screen)) {
const day =
screen.screen === 'stock' || screen.screen === 'bond'
? moscowDate(now)
: now.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH);
const from =
screen.screen === 'events' && screen.from
? new Date(`${screen.from}T00:00:00Z`)
: new Date(`${day}T00:00:00Z`);
const to =
screen.screen === 'events' && screen.to
? new Date(`${screen.to}T00:00:00Z`)
: new Date(from);
if (
!(screen.screen === 'events' && screen.to) &&
screen.screen !== 'stock' &&
screen.screen !== 'bond'
) {
to.setUTCDate(to.getUTCDate() + defaultEventDays);
}
if (screen.screen === 'stock' || screen.screen === 'bond') {
from.setUTCFullYear(from.getUTCFullYear() - incomeHistoryYears);
to.setUTCFullYear(to.getUTCFullYear() + FORECAST_YEARS);
}
sources.push({
...identity,
kind: 'calendar',
from,
to,
priority: 'screen',
});
}
}
if (!force && !['bonds', 'stocks'].includes(screen.screen)) {
sources.push(
...accountHistory(accounts, now).map((range) => ({
...range,
kind: 'operations' as const,
priority: 'history' as const,
})),
);
}
const priorities = { discovery: 0, screen: 1, history: 2 };
sources.sort(
(left, right) => priorities[left.priority] - priorities[right.priority],
);
return { sources, unavailable, now };
}
export function accountHistory(
accounts: Account[],
src/modules/investments/operations/application/brokerage-operations-sync.service.ts
-27
View File
@@ -6,13 +6,11 @@ import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
import { TbankOperationState } from '../../../../integrations/tbank/tbank-protocol';
import { TbankAccessService, TbankClient } from '../../../../integrations/tbank/tbank.api';
import { accountHistory } from '../../investment-sync-plan';
import {
type OperationInterval,
operationIntervals,
operationsSyncPriority,
planOperationRead,
uncoveredOperationIntervals,
} from '../domain/operations-sync-plan';
import {
BrokerageOperationsRepository,
@@ -48,7 +46,6 @@ export class BrokerageOperationsSyncService {
intervals: OperationInterval[],
options: {
force?: boolean;
history?: boolean;
movingEnd?: boolean;
observe?: boolean;
revision?: string;
@@ -99,30 +96,6 @@ export class BrokerageOperationsSyncService {
connection.revision,
)
: null;
if (!options.observe && !options.force && options.history) {
const accountIds = new Set(
intervals.map((interval) => interval.accountId),
);
const history = accountHistory(
connection.accounts.filter((account) =>
accountIds.has(account.accountId),
),
now,
);
const missing = planOperationRead(history, coverage, now, {
movingEnd: true,
}).load;
const remaining = uncoveredOperationIntervals(missing, targets);
if (remaining.length) {
await this.request(
userId,
remaining,
false,
'history',
connection.revision,
);
}
}
return { plan, handle };
}
src/modules/investments/operations/application/brokerage-operations.service.ts
+7 -48
View File
@@ -6,17 +6,12 @@ import {
InstrumentMetadataService,
metadataKey,
TbankAccessService,
TbankMembershipService,
type TbankOperationItem,
tbankSourceTtlMs,
} from '../../../../integrations/tbank/tbank.api';
import { dateOnlyBounds, DateOnlyRange } from '../../investment-calendar';
import { investmentResourceMeta } from '../../investment-resource.mapper';
import {
accountHistory,
ledgerRange,
planInvestmentScreen,
} from '../../investment-sync-plan';
import { accountHistory, ledgerRange } from '../../investment-sync-plan';
import {
hourlyAttempt,
TbankSyncRepository,
@@ -37,7 +32,6 @@ import {
} from '../presentation/brokerage-operation.dto';
import { BrokerageOperationsSyncService } from './brokerage-operations-sync.service';
const EXCLUSIVE_END_OFFSET_MS = 1;
const FIRST_RESULT_OFFSET = 0;
const MISSING_MEMBERSHIP_REVISION = 0;
const STALE_FRESHNESS_MS = 0;
@@ -61,7 +55,6 @@ export class BrokerageOperationsService {
private readonly metadata: InstrumentMetadataService,
private readonly repository: BrokerageOperationsRepository,
private readonly sync: BrokerageOperationsSyncService,
private readonly membership: TbankMembershipService,
private readonly config: ConfigService,
private readonly hourlySync: TbankSyncRepository,
) {}
@@ -117,35 +110,14 @@ export class BrokerageOperationsService {
cursor,
};
if (cursor) await this.repository.readPage(readInput);
const screen = planInvestmentScreen(
{
screen: 'operations',
accountId,
from: from.toISOString(),
to: new Date(to.valueOf() - EXCLUSIVE_END_OFFSET_MS).toISOString(),
},
{
accounts: selected,
instruments: [],
membershipKnown: Boolean(connection.membershipAt),
},
now,
);
const intervals = screen.sources.flatMap((source) =>
source.kind === 'operations' &&
source.priority === 'screen' &&
source.accountId &&
source.from &&
source.to
? [{ accountId: source.accountId, from: source.from, to: source.to }]
: [],
const intervals = selected.flatMap((account) =>
from < to ? [{ accountId: account.accountId, from, to }] : [],
);
// The ledger only starts the first-time load of missing intervals; the
// mutable tail belongs to the hourly walk and membership is not rechecked.
if (intervals.length) {
await this.sync.readPlan(userId, intervals, {
movingEnd,
history: true,
revision: connection.revision,
});
}
@@ -406,32 +378,19 @@ export class BrokerageOperationsService {
periods: ranges,
};
let saved = await this.repository.readPage(input);
const schedule = async () => {
const current = await this.access.local(userId, accountId);
if (current.revision !== connection.revision) return;
const snapshot = await this.repository.readPage(input);
if (!snapshot?.connection.membershipAt) return;
const targets = snapshot.connection.accounts.flatMap((account) =>
(snapshot.periods ?? []).map((period) => ({
if (!observe) {
const targets = (saved?.connection.accounts ?? []).flatMap((account) =>
(saved?.periods ?? []).map((period) => ({
accountId: account.accountId,
...period,
})),
);
if (targets.length) {
await this.sync.readPlan(userId, targets, {
movingEnd: snapshot.periods?.some((period) => period.movingEnd),
history: true,
movingEnd: saved?.periods?.some((period) => period.movingEnd),
revision: connection.revision,
});
}
};
if (!observe) {
const discovery = await this.membership.request(
userId,
connection.revision,
);
if (discovery) void discovery.done.then(schedule).catch(() => undefined);
else await schedule();
saved = await this.repository.readPage(input);
}
const now = new Date();
test/tbank-income-local.e2e-spec.ts
+52
View File
@@ -297,6 +297,58 @@ describe('local income completeness and exact money (SQLite)', () => {
expect(result.data?.calculatedAt).toBe('2025-02-01T00:00:00.000Z');
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
});
it('ordinary covered income reads local data without membership or operations RPC', async () => {
await prisma.tbankConnection.update({
where: { userId: 'income-owner' },
data: { membershipAt: new Date('2020-01-01Z') },
});
const response = await get('a', undefined, false).expect(200);
const result = response.body as IncomeAnalyticsResourceDto;
expect(result.data).not.toBeNull();
expect(result.meta.completeness).toBe('complete');
expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
});
it('cold income imports only the missing requested interval', async () => {
await prisma.tbankConnection.update({
where: { userId: 'income-owner' },
data: { membershipAt: new Date('2020-01-01Z') },
});
let release: () => void = () => {
throw new Error('Upstream not initialized');
};
const blocked = new Promise<{ hasNext: boolean; items: never[] }>((resolve) => {
release = () => resolve({ hasNext: false, items: [] });
});
upstream.getOperationsByCursor.mockReturnValue(blocked);
const first = await get('a', 'from=2025-02-01&to=2025-02-28', false).expect(200);
const cold = first.body as IncomeAnalyticsResourceDto;
release();
expect(cold.data).toBeNull();
expect(cold.meta.completeness).toBe('incomplete');
let result: IncomeAnalyticsResourceDto = cold;
for (let attempt = 0; attempt < 100; attempt++) {
const response = await get('a', 'from=2025-02-01&to=2025-02-28').expect(200);
result = response.body as IncomeAnalyticsResourceDto;
if (result.meta.completeness === 'complete') break;
await new Promise((resolve) => setTimeout(resolve, 10));
}
expect(result.data).not.toBeNull();
expect(result.meta.completeness).toBe('complete');
expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
expect(upstream.getOperationsByCursor.mock.calls).toEqual([
['synthetic', expect.objectContaining({
accountId: 'a',
from: { seconds: '1738357199' },
to: { seconds: '1738368001' },
})],
['synthetic', expect.objectContaining({
accountId: 'a',
from: { seconds: '1738367999' },
to: { seconds: '1740776401' },
})],
]);
});
it('returns while a cold upstream is blocked and validates access/period before scheduling', async () => {
await get('absent', undefined, false).expect(404);
await get('a', 'from=2025-02-31&to=2025-03-01', false).expect(400);
test/tbank-ledger-local.e2e-spec.ts
+45 -1
View File
@@ -113,6 +113,50 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
}
}
});
it.each([
'from=2099-01-01',
'from=2025-03-01&to=2025-02-01',
])('empty ledger range %s returns 200 without upstream RPC', async (query) => {
const response = await request(app.getHttpServer())
.get(`/api/v1/brokerage-accounts/a/operations?${query}`)
.set('Authorization', auth())
.expect(200);
const result = response.body as BrokerageOperationsResourceDto;
expect(result.data?.items).toEqual([]);
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
});
it('cold ledger imports only the missing requested interval', async () => {
upstream.getOperationsByCursor.mockResolvedValue({
hasNext: false,
items: [],
});
await put().expect(200);
await request(app.getHttpServer())
.get('/api/v1/brokerage-accounts/a/operations?from=2025-02-01&to=2025-02-28')
.set('Authorization', auth())
.expect(200);
let complete = false;
for (let attempt = 0; attempt < 100; attempt++) {
const response = await request(app.getHttpServer())
.get('/api/v1/brokerage-accounts/a/operations?from=2025-02-01&to=2025-02-28&sync=observe')
.set('Authorization', auth())
.expect(200);
const result = response.body as BrokerageOperationsResourceDto;
complete = result.meta.completeness === 'complete';
if (complete) break;
await delay(10);
}
expect(complete).toBe(true);
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
expect(upstream.getOperationsByCursor).toHaveBeenCalledWith(
'synthetic',
expect.objectContaining({
accountId: 'a',
from: { seconds: '1738367999' },
to: { seconds: '1740787201' },
}),
);
});
it('reads ordered pages with identical IDs/timestamps across accounts and no upstream RPC, including after reopen', async () => {
let cursor: null | string = null;
const ids: string[] = [];
@@ -409,7 +453,7 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
const result = (await get('sync=observe', 'a').expect(200))
.body as BrokerageOperationsResourceDto;
expect(result.meta.coverage?.historyBoundary).toBe('unknown');
expect(result.meta.coverage?.historyState).toBe('loading');
expect(result.meta.coverage?.historyState).toBe('missing');
}, 30_000);
it('returns cold missing while upstream is blocked, then confirms an empty interval; observe never retries failure', async () => {