feat: reconcile hourly investment resource states

src/modules/investments/portfolio/application/forecast.service.ts

@@ -2,6 +2,7 @@ import { Injectable } from '@nestjs/common';
 
 import type { InvestmentResourceMetaDto } from '../../investment-resource.dto';
 
+import { add, decimalFromString, formatDecimal } from '../../../../common/decimal';
 import { investmentResourceMeta } from '../../investment-resource.mapper';
 import {
   ForecastRepository,
@@ -16,6 +17,7 @@ export class ForecastService {
 
   async result<T extends object>(input: {
     candidate?: T;
+    lastAttempt?: InvestmentResourceMetaDto['lastAttempt'];
     membershipRevision: number;
     portfolio: InvestmentResourceMetaDto;
     revision: string;
@@ -72,22 +74,51 @@ export class ForecastService {
           .map((dependency) => dependency.sync)
           .find((sync) => sync.state === 'failed');
     return {
-      data: stored ? (stored.payload as T) : null,
+      data: stored ? filterEvents(stored.payload as T, input.scopeKey) : null,
       meta: investmentResourceMeta({
         resourceKey: input.scopeKey,
         dataRevision: String(stored?.dataRevision ?? MISSING_DATA_REVISION),
         availability: stored
           ? 'complete'
           : dependencies.some(
-            (dependency) => dependency.availability !== 'missing',
+            (dependency) => dependency.completeness === 'complete',
           )
             ? 'partial'
             : 'missing',
         fetchedAt: first ? new Date(first) : null,
         freshForMs: Number.POSITIVE_INFINITY,
         work: work?.id ? { ...work, id: work.id } : null,
+        lastAttempt: input.lastAttempt,
         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;
+  }
+}

src/modules/investments/portfolio/application/portfolio.service.spec.ts

@@ -26,6 +26,7 @@ describe('persistent portfolio projection', () => {
       repository as never,
       { request: jest.fn().mockResolvedValue(null) } as never,
       { event: jest.fn() } as never,
+      { hourly: jest.fn().mockResolvedValue(null) } as never,
     );
     access.local.mockResolvedValue({
       revision: 'private',
@@ -192,7 +193,7 @@ describe('persistent portfolio projection', () => {
     });
     const partial = await service.read('user', 'all');
     expect(partial.data).toBeNull();
-    expect(partial.meta.availability).toBe('partial');
+    expect(partial.meta.availability).toBe('temporarily-unavailable');
     const saved = await service.saved('user', 'all');
     expect(saved.accounts[0].snapshot?.totalsByCurrency).toEqual(
       snapshot.totalsByCurrency,

src/modules/investments/portfolio/application/portfolio.service.ts

@@ -17,6 +17,10 @@ import {
   tbankSourceTtlMs,
 } from '../../../../integrations/tbank/tbank.api';
 import { investmentResourceMeta } from '../../investment-resource.mapper';
+import {
+  hourlyAttempt,
+  TbankSyncRepository,
+} from '../../tbank-sync/tbank-sync.repository';
 import { portfolioStructure } from '../domain/portfolio-structure';
 import { sumMoneyByCurrency } from '../domain/portfolio-totals';
 import {
@@ -44,12 +48,14 @@ export class PortfolioService {
     private readonly repository: PortfolioRepository,
     private readonly membership: TbankMembershipService,
     private readonly logger: StructuredLogger,
+    private readonly sync: TbankSyncRepository,
   ) {}
 
   // Local reads never start upstream work; the hourly walk owns refreshes.
   async saved(userId: string, accountId: string) {
     const connection = await this.access.local(userId, accountId);
     const saved = await this.repository.read(userId, connection.revision);
+    const hourly = await this.sync.hourly(userId);
     if (
       saved?.membershipAt &&
       accountId !== 'all' &&
@@ -64,8 +70,10 @@ export class PortfolioService {
       fetchedAt: saved?.membershipAt ?? null,
       freshForMs: membershipFreshForMs,
       work: saved?.work.find((work) => work.scopeKey === 'membership') ?? null,
+      lastAttempt: hourlyAttempt(hourly, 'membership'),
     });
     return {
+      hourly,
       membership: membershipMeta,
       accounts: (saved?.accounts ?? [])
         .filter(
@@ -95,15 +103,16 @@ export class PortfolioService {
               saved?.work.find(
                 (work) => work.scopeKey === portfolioWorkKey(account.accountId),
               ) ?? null,
+            lastAttempt: hourlyAttempt(hourly, 'portfolio', account.accountId),
           }),
         })),
     };
   }
 
   async read(userId: string, accountId: string) {
-    const saved = await this.saved(userId, accountId);
+    const { hourly, ...saved } = await this.saved(userId, accountId);
     const missing =
-      saved.membership.availability === 'missing' ||
+      saved.membership.completeness === 'incomplete' ||
       saved.accounts.some((account) => !account.snapshot);
     const metadata = await this.metadata.savedMany(
       saved.accounts.flatMap((account) => account.snapshot?.positions ?? []),
@@ -206,6 +215,7 @@ export class PortfolioService {
           ? portfolioFreshForMs
           : membershipFreshForMs,
         work: work.id ? { ...work, id: work.id } : null,
+        lastAttempt: hourlyAttempt(hourly, 'portfolio', accountId === 'all' ? undefined : accountId),
         dependencies: [
           saved.membership,
           ...saved.accounts.map((account) => account.meta),

src/modules/investments/portfolio/infrastructure/forecast.repository.ts

@@ -51,11 +51,32 @@ export class ForecastRepository {
       const previous = await tx.tbankForecast.findUnique({
         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 =
-        previous?.membershipRevision === input.membershipRevision
-          ? previous
+        saved?.membershipRevision === input.membershipRevision
+          ? saved
           : null;
       if (!input.candidate) return fallback;
+      if (!previous && fallback) return fallback;
       const portfolios = await tx.tbankPortfolio.findMany({
         where: {
           ...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;
+  }
+}

src/modules/investments/portfolio/portfolio.module.ts

@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 
 import { TbankModule } from '../../../integrations/tbank/tbank.module';
 import { AuthModule } from '../../auth/auth.module';
+import { TbankSyncRepositoryModule } from '../tbank-sync/tbank-sync-repository.module';
 import { ForecastService } from './application/forecast.service';
 import { PortfolioService } from './application/portfolio.service';
 import { ForecastRepository } from './infrastructure/forecast.repository';
@@ -9,7 +10,7 @@ import { PortfolioRepository } from './infrastructure/portfolio.repository';
 import { PortfolioController } from './presentation/portfolio.controller';
 
 @Module({
-  imports: [AuthModule, TbankModule],
+  imports: [AuthModule, TbankModule, TbankSyncRepositoryModule],
   controllers: [PortfolioController],
   providers: [
     PortfolioService,