fix(portfolio): publish new complete forecast regardless of request window (WealthLedger/backend#55) #156
@@ -20,6 +20,7 @@ export class ForecastService {
|
|||||||
lastAttempt?: InvestmentResourceMetaDto['lastAttempt'];
|
lastAttempt?: InvestmentResourceMetaDto['lastAttempt'];
|
||||||
membershipRevision: number;
|
membershipRevision: number;
|
||||||
portfolio: InvestmentResourceMetaDto;
|
portfolio: InvestmentResourceMetaDto;
|
||||||
|
replacesLatest?: boolean;
|
||||||
revision: string;
|
revision: string;
|
||||||
scopeKey: string;
|
scopeKey: string;
|
||||||
sources: {
|
sources: {
|
||||||
@@ -74,7 +75,7 @@ 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 ? filterEvents(stored.payload as T, input.scopeKey) : null,
|
data: stored ? filterEvents(stored.payload as T, input.scopeKey, stored.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),
|
||||||
@@ -95,7 +96,7 @@ export class ForecastService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
function filterEvents<T extends object>(payload: T, scopeKey: string): T {
|
function filterEvents<T extends object>(payload: T, scopeKey: string, savedScopeKey: string): T {
|
||||||
if (!scopeKey.startsWith('events/')) return payload;
|
if (!scopeKey.startsWith('events/')) return payload;
|
||||||
try {
|
try {
|
||||||
const [, from, to] = JSON.parse(scopeKey.slice('events/'.length)) as [string, string, string];
|
const [, from, to] = JSON.parse(scopeKey.slice('events/'.length)) as [string, string, string];
|
||||||
@@ -111,8 +112,10 @@ function filterEvents<T extends object>(payload: T, scopeKey: string): T {
|
|||||||
const current = totals.get(event.projectedAmount.currency) ?? { value: 0n, scale: 0 };
|
const current = totals.get(event.projectedAmount.currency) ?? { value: 0n, scale: 0 };
|
||||||
totals.set(event.projectedAmount.currency, add(current, decimalFromString(event.projectedAmount.value)));
|
totals.set(event.projectedAmount.currency, add(current, decimalFromString(event.projectedAmount.value)));
|
||||||
}
|
}
|
||||||
|
const [, savedFrom, savedTo] = JSON.parse(savedScopeKey.slice('events/'.length)) as [string, string, string];
|
||||||
return {
|
return {
|
||||||
...value,
|
...value,
|
||||||
|
window: { from: savedFrom, to: savedTo },
|
||||||
events,
|
events,
|
||||||
incomeTotalsByCurrency: [...totals]
|
incomeTotalsByCurrency: [...totals]
|
||||||
.sort(([left], [right]) => left.localeCompare(right))
|
.sort(([left], [right]) => left.localeCompare(right))
|
||||||
|
|||||||
@@ -26,6 +26,7 @@ export class ForecastRepository {
|
|||||||
membershipRevision: number;
|
membershipRevision: number;
|
||||||
portfolioProjection: string;
|
portfolioProjection: string;
|
||||||
portfolios: { accountId: string; dataRevision: number }[];
|
portfolios: { accountId: string; dataRevision: number }[];
|
||||||
|
replacesLatest?: boolean;
|
||||||
revision: string;
|
revision: string;
|
||||||
scopeKey: string;
|
scopeKey: string;
|
||||||
sources: ForecastSourceVersion[];
|
sources: ForecastSourceVersion[];
|
||||||
@@ -51,32 +52,30 @@ 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 },
|
||||||
});
|
});
|
||||||
|
// An events read serves the account's latest complete forecast whatever
|
||||||
|
// the requested window; other scopes keep their exact-key row.
|
||||||
const requestedWindow = eventsWindow(input.scopeKey);
|
const requestedWindow = eventsWindow(input.scopeKey);
|
||||||
const saved = previous ?? (requestedWindow
|
const latest = requestedWindow
|
||||||
? (
|
? await tx.tbankForecast.findFirst({
|
||||||
await tx.tbankForecast.findMany({
|
where: {
|
||||||
where: {
|
...owner,
|
||||||
...owner,
|
// scopeKey is `events/[accountId,from,to]` JSON: pin the account.
|
||||||
scopeKey: { startsWith: 'events/' },
|
scopeKey: {
|
||||||
membershipRevision: input.membershipRevision,
|
startsWith: `events/[${JSON.stringify(requestedWindow.accountId)},`,
|
||||||
},
|
},
|
||||||
})
|
membershipRevision: input.membershipRevision,
|
||||||
).find((forecast) => {
|
},
|
||||||
const window = eventsWindow(forecast.scopeKey);
|
orderBy: [{ calculatedAt: 'desc' }, { scopeKey: 'asc' }],
|
||||||
return (
|
|
||||||
window &&
|
|
||||||
window.accountId === requestedWindow.accountId &&
|
|
||||||
window.from <= requestedWindow.to &&
|
|
||||||
requestedWindow.from <= window.to
|
|
||||||
);
|
|
||||||
})
|
})
|
||||||
: null);
|
: previous;
|
||||||
const fallback =
|
const fallback =
|
||||||
saved?.membershipRevision === input.membershipRevision
|
latest?.membershipRevision === input.membershipRevision ? latest : null;
|
||||||
? saved
|
|
||||||
: null;
|
|
||||||
if (!input.candidate) return fallback;
|
if (!input.candidate) return fallback;
|
||||||
if (!previous && fallback) return fallback;
|
// Only the default window may replace the latest forecast; an explicit
|
||||||
|
// range is served from it and publishes only when nothing is saved.
|
||||||
|
if (!input.replacesLatest && fallback && fallback.scopeKey !== input.scopeKey) {
|
||||||
|
return fallback;
|
||||||
|
}
|
||||||
const portfolios = await tx.tbankPortfolio.findMany({
|
const portfolios = await tx.tbankPortfolio.findMany({
|
||||||
where: {
|
where: {
|
||||||
...owner,
|
...owner,
|
||||||
@@ -152,7 +151,7 @@ export class ForecastRepository {
|
|||||||
) as Prisma.InputJsonObject;
|
) as Prisma.InputJsonObject;
|
||||||
const timestamps = input.timestamps as Prisma.InputJsonArray;
|
const timestamps = input.timestamps as Prisma.InputJsonArray;
|
||||||
if (
|
if (
|
||||||
fallback &&
|
fallback?.scopeKey === input.scopeKey &&
|
||||||
isDeepStrictEqual(fallback.dependencyRevisions, revisions) &&
|
isDeepStrictEqual(fallback.dependencyRevisions, revisions) &&
|
||||||
isDeepStrictEqual(fallback.sourceTimestamps, timestamps)
|
isDeepStrictEqual(fallback.sourceTimestamps, timestamps)
|
||||||
) {
|
) {
|
||||||
|
|||||||
@@ -65,6 +65,14 @@ export class UpcomingEventDto {
|
|||||||
declare projectedAmount: MoneyDto | null;
|
declare projectedAmount: MoneyDto | null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export class UpcomingEventsWindowDto {
|
||||||
|
@ApiProperty({ format: 'date' })
|
||||||
|
declare from: string;
|
||||||
|
|
||||||
|
@ApiProperty({ format: 'date' })
|
||||||
|
declare to: string;
|
||||||
|
}
|
||||||
|
|
||||||
export class UpcomingEventsDto {
|
export class UpcomingEventsDto {
|
||||||
@ApiProperty()
|
@ApiProperty()
|
||||||
declare accountId: string;
|
declare accountId: string;
|
||||||
@@ -72,6 +80,13 @@ export class UpcomingEventsDto {
|
|||||||
@ApiProperty({ format: 'date-time' })
|
@ApiProperty({ format: 'date-time' })
|
||||||
declare calculatedAt: string;
|
declare calculatedAt: string;
|
||||||
|
|
||||||
|
@ApiProperty({
|
||||||
|
type: UpcomingEventsWindowDto,
|
||||||
|
description:
|
||||||
|
'Inclusive UTC window the saved forecast was calculated for. Unchanged by the requested from/to filter.',
|
||||||
|
})
|
||||||
|
declare window: UpcomingEventsWindowDto;
|
||||||
|
|
||||||
@ApiProperty({ type: UpcomingEventDto, isArray: true })
|
@ApiProperty({ type: UpcomingEventDto, isArray: true })
|
||||||
declare events: UpcomingEventDto[];
|
declare events: UpcomingEventDto[];
|
||||||
|
|
||||||
|
|||||||
@@ -150,7 +150,7 @@ export class UpcomingEventsService {
|
|||||||
a.type.localeCompare(b.type) ||
|
a.type.localeCompare(b.type) ||
|
||||||
a.ticker.localeCompare(b.ticker),
|
a.ticker.localeCompare(b.ticker),
|
||||||
);
|
);
|
||||||
const candidate: undefined | UpcomingEventsDto = ready
|
const candidate: Omit<UpcomingEventsDto, 'window'> | undefined = ready
|
||||||
? {
|
? {
|
||||||
accountId,
|
accountId,
|
||||||
calculatedAt: new Date().toISOString(),
|
calculatedAt: new Date().toISOString(),
|
||||||
@@ -163,6 +163,7 @@ export class UpcomingEventsService {
|
|||||||
revision: connection.revision,
|
revision: connection.revision,
|
||||||
membershipRevision: connection.membershipRevision,
|
membershipRevision: connection.membershipRevision,
|
||||||
scopeKey: `events/${JSON.stringify([accountId, window.from, window.to])}`,
|
scopeKey: `events/${JSON.stringify([accountId, window.from, window.to])}`,
|
||||||
|
replacesLatest: query.from === undefined && query.to === undefined,
|
||||||
portfolio: portfolio.meta,
|
portfolio: portfolio.meta,
|
||||||
lastAttempt: hourlyAttempt(
|
lastAttempt: hourlyAttempt(
|
||||||
hourly,
|
hourly,
|
||||||
|
|||||||
@@ -82,7 +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') => {
|
const refresh = async (
|
||||||
|
range: { from?: string; to?: string } = window,
|
||||||
|
accountId = 'account',
|
||||||
|
) => {
|
||||||
await app.get(UpcomingEventsService).refresh('owner', accountId, range);
|
await app.get(UpcomingEventsService).refresh('owner', accountId, range);
|
||||||
await settled();
|
await settled();
|
||||||
};
|
};
|
||||||
@@ -318,17 +321,23 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
|
|||||||
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('serves a disjoint range as an empty saved result and never relabels a membership snapshot', async () => {
|
||||||
await refresh();
|
await refresh();
|
||||||
await settled();
|
await settled();
|
||||||
await events().expect(200);
|
const saved = (await events().expect(200))
|
||||||
|
.body as UpcomingEventsResourceDto;
|
||||||
const newRange = (
|
const newRange = (
|
||||||
await events(true, 'account', 'owner', {
|
await events(true, 'account', 'owner', {
|
||||||
from: '2099-03-01',
|
from: '2099-03-01',
|
||||||
to: '2099-03-31',
|
to: '2099-03-31',
|
||||||
}).expect(200)
|
}).expect(200)
|
||||||
).body as UpcomingEventsResourceDto;
|
).body as UpcomingEventsResourceDto;
|
||||||
expect(newRange.data).toBeNull();
|
expect(newRange.data).toEqual({
|
||||||
|
...saved.data,
|
||||||
|
events: [],
|
||||||
|
incomeTotalsByCurrency: [],
|
||||||
|
});
|
||||||
|
expect(newRange.data?.window).toEqual(window);
|
||||||
await prisma.tbankConnection.update({
|
await prisma.tbankConnection.update({
|
||||||
where: { userId: 'owner' },
|
where: { userId: 'owner' },
|
||||||
data: { membershipRevision: { increment: 1 } },
|
data: { membershipRevision: { increment: 1 } },
|
||||||
@@ -640,6 +649,132 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => {
|
|||||||
expect(moex.bondEvents).not.toHaveBeenCalled();
|
expect(moex.bondEvents).not.toHaveBeenCalled();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('keeps the latest wide forecast when an explicit narrower range has a complete candidate', async () => {
|
||||||
|
await refresh();
|
||||||
|
const full = (await events().expect(200))
|
||||||
|
.body as UpcomingEventsResourceDto;
|
||||||
|
const narrowRange = { from: '2099-02-01', to: '2099-02-10' };
|
||||||
|
await refresh(narrowRange);
|
||||||
|
|
||||||
|
const narrow = (await events(true, 'account', 'owner', narrowRange).expect(
|
||||||
|
200,
|
||||||
|
)).body as UpcomingEventsResourceDto;
|
||||||
|
|
||||||
|
expect(narrow.data?.calculatedAt).toBe(full.data?.calculatedAt);
|
||||||
|
expect(narrow.data?.window).toEqual(window);
|
||||||
|
expect(narrow.data?.events.map((event) => event.date)).toEqual([
|
||||||
|
'2099-02-01',
|
||||||
|
]);
|
||||||
|
expect(await prisma.tbankForecast.count()).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('daily default window shift', () => {
|
||||||
|
const day0 = '2099-02-01';
|
||||||
|
const day1 = '2099-02-02';
|
||||||
|
const windowOf = (from: string, to: string) => ({ from, to });
|
||||||
|
const w0 = windowOf(day0, '2099-03-03');
|
||||||
|
const w1 = windowOf(day1, '2099-03-04');
|
||||||
|
const dividend = (date: string) => ({
|
||||||
|
recordDate: time(date),
|
||||||
|
dividendNet: { units: '10', nano: 1, currency: 'usd' },
|
||||||
|
});
|
||||||
|
const defaultEvents = async () =>
|
||||||
|
(await get('brokerage-accounts/account/events?sync=observe').expect(200))
|
||||||
|
.body as UpcomingEventsResourceDto;
|
||||||
|
// Only Date is faked: the app and SQLite helpers need real timers.
|
||||||
|
const clock = (date: string) => {
|
||||||
|
jest.useFakeTimers({
|
||||||
|
doNotFake: [
|
||||||
|
'hrtime', 'nextTick', 'performance', 'queueMicrotask',
|
||||||
|
'requestAnimationFrame', 'cancelAnimationFrame',
|
||||||
|
'requestIdleCallback', 'cancelIdleCallback', 'setImmediate',
|
||||||
|
'clearImmediate', 'setInterval', 'clearInterval', 'setTimeout',
|
||||||
|
'clearTimeout',
|
||||||
|
],
|
||||||
|
now: new Date(`${date}T12:00:00Z`),
|
||||||
|
});
|
||||||
|
};
|
||||||
|
afterEach(() => jest.useRealTimers());
|
||||||
|
|
||||||
|
it('publishes the new window as latest and serves it for stale or disjoint explicit ranges', async () => {
|
||||||
|
clock(day0);
|
||||||
|
upstream.getDividends.mockResolvedValue([
|
||||||
|
dividend('2099-02-05'),
|
||||||
|
dividend('2099-03-04'),
|
||||||
|
]);
|
||||||
|
await portfolio([position(shareUid, '2')]);
|
||||||
|
await refresh({});
|
||||||
|
const first = await defaultEvents();
|
||||||
|
expect(first.data?.window).toEqual(w0);
|
||||||
|
expect(first.data?.incomeTotalsByCurrency).toEqual([
|
||||||
|
{ currency: 'USD', value: '20.000000002' },
|
||||||
|
]);
|
||||||
|
|
||||||
|
jest.setSystemTime(new Date(`${day1}T12:00:00Z`));
|
||||||
|
await portfolio([position(shareUid, '5')]);
|
||||||
|
await refresh({});
|
||||||
|
const second = await defaultEvents();
|
||||||
|
|
||||||
|
expect(second.data?.window).toEqual(w1);
|
||||||
|
expect(second.data?.calculatedAt).not.toBe(first.data?.calculatedAt);
|
||||||
|
expect(second.data?.events.map((event) => event.date)).toEqual([
|
||||||
|
'2099-02-05',
|
||||||
|
'2099-03-04',
|
||||||
|
]);
|
||||||
|
expect(second.data?.incomeTotalsByCurrency).toEqual([
|
||||||
|
{ currency: 'USD', value: '100.00000001' },
|
||||||
|
]);
|
||||||
|
const stale = (await events(true, 'account', 'owner', w0).expect(200))
|
||||||
|
.body as UpcomingEventsResourceDto;
|
||||||
|
expect(stale.data?.calculatedAt).toBe(second.data?.calculatedAt);
|
||||||
|
expect(stale.data?.window).toEqual(w1);
|
||||||
|
const disjoint = (
|
||||||
|
await events(true, 'account', 'owner', {
|
||||||
|
from: '2100-01-01',
|
||||||
|
to: '2100-01-31',
|
||||||
|
}).expect(200)
|
||||||
|
).body as UpcomingEventsResourceDto;
|
||||||
|
expect(disjoint.data).toEqual({
|
||||||
|
...second.data,
|
||||||
|
events: [],
|
||||||
|
incomeTotalsByCurrency: [],
|
||||||
|
});
|
||||||
|
expect(await prisma.tbankForecast.count()).toBe(2);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('keeps the previous complete forecast when a holding calendar fails after the shift', async () => {
|
||||||
|
clock(day0);
|
||||||
|
upstream.getDividends.mockResolvedValue([dividend('2099-02-05')]);
|
||||||
|
await refresh({});
|
||||||
|
const before = await defaultEvents();
|
||||||
|
expect(before.data?.window).toEqual(w0);
|
||||||
|
|
||||||
|
jest.setSystemTime(new Date(`${day1}T12:00:00Z`));
|
||||||
|
await portfolio([position(shareUid, '7'), position(newUid, '5')]);
|
||||||
|
upstream.getDividends.mockImplementation((_token: string, uid: string) =>
|
||||||
|
uid === newUid
|
||||||
|
? Promise.reject(new Error('new calendar unavailable'))
|
||||||
|
: Promise.resolve([dividend('2099-02-05')]),
|
||||||
|
);
|
||||||
|
await refresh({});
|
||||||
|
|
||||||
|
expect((await defaultEvents()).data).toEqual(before.data);
|
||||||
|
expect(await prisma.tbankForecast.count()).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('publishes the shifted window even when no source depends on it', async () => {
|
||||||
|
clock(day0);
|
||||||
|
await portfolio([]);
|
||||||
|
await refresh({});
|
||||||
|
expect((await defaultEvents()).data?.window).toEqual(w0);
|
||||||
|
|
||||||
|
jest.setSystemTime(new Date(`${day1}T12:00:00Z`));
|
||||||
|
await refresh({});
|
||||||
|
|
||||||
|
expect((await defaultEvents()).data?.window).toEqual(w1);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
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(
|
||||||
|
|||||||
Reference in New Issue
Block a user