fix(events): refresh saved payout calendars once per hourly pass (WealthLedger/backend#53) (#157)

src/modules/investments/instruments/application/catalog-enrichment.service.ts

@@ -110,6 +110,7 @@ export class CatalogEnrichmentService {
       tickers?: Record<string, string>;
       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<string, string>;
       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) {

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' });

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<string[]> {
     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<ReturnType<CatalogEnrichmentService['readMany']>> =
@@ -229,6 +232,7 @@ export class UpcomingEventsService {
               identities.map((entry) => [entry.uid, entry.ticker]),
             ),
           },
+          refreshBefore,
         );
         sources.push(...batch);
         for (const source of batch) {

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(