From aeb4647fda3806a419ff3578834c34fff97ae3e8 Mon Sep 17 00:00:00 2001 From: codex-bot Date: Mon, 2 Nov 2026 23:06:14 +0300 Subject: [PATCH] fix(events): refresh saved payout calendars once per hourly pass (WealthLedger/backend#53) (#157) --- .../application/catalog-enrichment.service.ts | 8 +- .../tbank-sync/tbank-hourly.scheduler.ts | 6 +- .../upcoming-events.service.ts | 4 + test/upcoming-events.e2e-spec.ts | 115 ++++++++++++++++++ 4 files changed, 131 insertions(+), 2 deletions(-) diff --git a/src/modules/investments/instruments/application/catalog-enrichment.service.ts b/src/modules/investments/instruments/application/catalog-enrichment.service.ts index 1bfa0ae..217a77b 100644 --- a/src/modules/investments/instruments/application/catalog-enrichment.service.ts +++ b/src/modules/investments/instruments/application/catalog-enrichment.service.ts @@ -110,6 +110,7 @@ export class CatalogEnrichmentService { tickers?: Record; to: string; }, + refreshBefore?: Date, ) { const { revision } = await this.access.local(userId); const inputs = sourceInputs(instrumentType, uids, kinds, range); @@ -123,6 +124,7 @@ export class CatalogEnrichmentService { false, kinds, range, + refreshBefore, ); } const work = inputs.length @@ -175,6 +177,7 @@ export class CatalogEnrichmentService { tickers?: Record; to: string; }, + refreshBefore?: Date, ) { const inputs = sourceInputs(instrumentType, uids, kinds, range); const reservation = this.reservations.then(async () => { @@ -189,7 +192,9 @@ export class CatalogEnrichmentService { ); const policy = sourcePolicy(input.kind); const needsRefresh = force || !row?.fetchedAt || - (policy === 'quote' && tbankSourceNeedsRefresh('quote', row.fetchedAt, new Date())); + (policy === 'quote' && tbankSourceNeedsRefresh('quote', row.fetchedAt, new Date())) || + // A walk reads a saved calendar once: later accounts see the new fetchedAt. + (policy === 'calendar' && refreshBefore !== undefined && row.fetchedAt < refreshBefore); if (!needsRefresh) { continue; } @@ -665,6 +670,7 @@ function sourcePolicy(kind: CatalogSource['kind']) { ) { return 'quote'; } + if (['bond-events', 'dividends'].includes(kind)) return 'calendar'; return 'stable'; } function producerKey(input: CatalogSource) { diff --git a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts index 1ac7b2a..058700d 100644 --- a/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts +++ b/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts @@ -268,6 +268,7 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { account.accountId, dispatch, dispatchProblems, + walkStartedAt, ); } } @@ -340,9 +341,12 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { accountId: string, dispatch: DispatchStep, dispatchProblems: HourlySection[], + walkStartedAt: Date, ) { try { - const ids = await this.withBound(this.events.refresh(userId, accountId)); + const ids = await this.withBound( + this.events.refresh(userId, accountId, {}, walkStartedAt), + ); if (ids !== BOUND_EXPIRED) { for (const id of ids) { await dispatch({ accountId, id, section: 'events' }); diff --git a/src/modules/investments/upcoming-events/upcoming-events.service.ts b/src/modules/investments/upcoming-events/upcoming-events.service.ts index 1e04ca6..67690a3 100644 --- a/src/modules/investments/upcoming-events/upcoming-events.service.ts +++ b/src/modules/investments/upcoming-events/upcoming-events.service.ts @@ -185,6 +185,7 @@ export class UpcomingEventsService { userId: string, accountId: string, query: UpcomingEventsQueryDto = {}, + walkStartedAt?: Date, ): Promise { const window = dateWindow(query); await this.access.local(userId, accountId); @@ -195,6 +196,7 @@ export class UpcomingEventsService { positions, window, false, + walkStartedAt, ); return launched; } @@ -204,6 +206,7 @@ export class UpcomingEventsService { positions: Position[], window: { from: string; to: string }, observe: boolean, + refreshBefore?: Date, ) { const metadata = await this.metadata.savedMany(positions); const sources: Awaited> = @@ -229,6 +232,7 @@ export class UpcomingEventsService { identities.map((entry) => [entry.uid, entry.ticker]), ), }, + refreshBefore, ); sources.push(...batch); for (const source of batch) { diff --git a/test/upcoming-events.e2e-spec.ts b/test/upcoming-events.e2e-spec.ts index 9272a3a..361f329 100644 --- a/test/upcoming-events.e2e-spec.ts +++ b/test/upcoming-events.e2e-spec.ts @@ -775,6 +775,121 @@ describe('saved payout calendars and complete forecasts (SQLite HTTP)', () => { }); }); + describe('hourly pass refreshes saved payout calendars', () => { + const date = new Date(Date.now() + 7 * 86400000).toISOString().slice(0, 10); + const dividend = (units: string, dividendType = 'Regular') => ({ + recordDate: time(date), + dividendType, + dividendNet: { units, currency: 'usd' }, + }); + // One hourly walk: the scheduler stamps it once and passes it to every account. + const startWalk = async () => { + await delay(10); + return new Date(); + }; + const pass = async (walkStartedAt: Date, accountId = 'account') => { + await app + .get(UpcomingEventsService) + .refresh('owner', accountId, {}, walkStartedAt); + await settled(); + }; + const latest = async () => + (await get('brokerage-accounts/account/events?sync=observe').expect(200)) + .body as UpcomingEventsResourceDto; + const dividendRow = () => + prisma.tbankInstrumentSource.findFirstOrThrow({ + where: { instrumentUid: shareUid, kind: 'dividends' }, + }); + + it('picks up a changed and then a cancelled payout on the next pass of the same window', async () => { + upstream.getDividends.mockResolvedValue([dividend('10')]); + await pass(await startWalk()); + expect((await latest()).data?.incomeTotalsByCurrency).toEqual([ + { currency: 'USD', value: '20' }, + ]); + const first = await dividendRow(); + + upstream.getDividends.mockResolvedValue([dividend('30')]); + await pass(await startWalk()); + expect((await latest()).data?.incomeTotalsByCurrency).toEqual([ + { currency: 'USD', value: '60' }, + ]); + expect((await dividendRow()).dataRevision).toBe(first.dataRevision + 1); + + upstream.getDividends.mockResolvedValue([dividend('30', 'Cancelled')]); + await pass(await startWalk()); + expect((await latest()).data?.events).toEqual([]); + }); + + it('keeps the first complete calendar, time and forecast when the next refresh fails', async () => { + upstream.getDividends.mockResolvedValue([dividend('10')]); + await pass(await startWalk()); + const before = await latest(); + const row = await dividendRow(); + + upstream.getDividends.mockRejectedValue(new Error('outage')); + await pass(await startWalk()); + const after = await latest(); + + expect(after.data).toEqual(before.data); + expect(after.meta.lastSuccessfulAt).toBe(before.meta.lastSuccessfulAt); + expect( + after.meta.dependencies?.some( + (dependency) => dependency.lastAttempt.state === 'failed', + ), + ).toBe(true); + const kept = await dividendRow(); + expect(kept.dataRevision).toBe(row.dataRevision); + expect(kept.fetchedAt).toEqual(row.fetchedAt); + }); + + it('loads one shared calendar once per walk across accounts and leaves detail, GET and a plain refresh alone', async () => { + await prisma.tbankAccount.create({ + data: { + userId: 'owner', + revision: 'owner-revision', + accountId: 'second', + type: 'ACCOUNT_TYPE_TINKOFF', + name: 'Second', + upstreamOrder: 1, + }, + }); + await portfolio([position()], 'owner', 'second'); + upstream.getDividends.mockResolvedValue([dividend('10')]); + await pass(await startWalk()); + const detailCalls = () => + upstream.getShare.mock.calls.length + + upstream.getInstrumentByUid.mock.calls.length; + const detailBefore = detailCalls(); + expect(upstream.getDividends).toHaveBeenCalledTimes(1); + + const walk = await startWalk(); + await pass(walk, 'account'); + await pass(walk, 'second'); + expect(upstream.getDividends).toHaveBeenCalledTimes(2); + + await app.get(UpcomingEventsService).refresh('owner', 'account', {}); + await settled(); + await latest(); + await get('brokerage-accounts/all/events').expect(200); + expect(upstream.getDividends).toHaveBeenCalledTimes(2); + expect(detailCalls()).toBe(detailBefore); + }); + + it('re-reads the bond calendar whose key never changes, only on a walk pass', async () => { + await portfolio([position(bondUid, '3', 'BOND')]); + await pass(await startWalk()); + expect(moex.bondEvents).toHaveBeenCalledTimes(1); + + await app.get(UpcomingEventsService).refresh('owner', 'account', {}); + await settled(); + expect(moex.bondEvents).toHaveBeenCalledTimes(1); + + await pass(await startWalk()); + expect(moex.bondEvents).toHaveBeenCalledTimes(2); + }); + }); + it('discards a late calendar from a replaced connection and prevents access to old private forecasts', async () => { let release: (value: unknown[]) => void = () => {}; upstream.getDividends.mockReturnValueOnce(