From 019bb01598d7ff358b2d4fe5eddc91b2b6afa3d0 Mon Sep 17 00:00:00 2001 From: Andrew Zolotukhin Date: Tue, 16 Jun 2026 04:22:29 +0000 Subject: [PATCH 1/5] feat: add cash-flow forecast report --- apps/api/src/api/endpoints.ts | 13 +- apps/api/src/api/handlers/index.ts | 4 +- apps/api/src/api/handlers/transactions.ts | 8 + .../application/cash-flow-forecast.test.ts | 302 +++++++ .../api/src/application/cash-flow-forecast.ts | 814 ++++++++++++++++++ apps/web/app/(app)/stats/forecast/page.tsx | 42 + .../cash-flow-forecast-explorer.test.tsx | 139 +++ .../cash-flow-forecast-explorer.tsx | 468 ++++++++++ apps/web/components/stats-explorer.tsx | 36 +- packages/contracts/src/api.test.ts | 3 +- packages/contracts/src/api.ts | 13 + packages/contracts/src/schemas.ts | 264 ++++++ tests/e2e/workflows.spec.ts | 17 + 13 files changed, 2108 insertions(+), 15 deletions(-) create mode 100644 apps/api/src/application/cash-flow-forecast.test.ts create mode 100644 apps/api/src/application/cash-flow-forecast.ts create mode 100644 apps/web/app/(app)/stats/forecast/page.tsx create mode 100644 apps/web/components/cash-flow-forecast-explorer.test.tsx create mode 100644 apps/web/components/cash-flow-forecast-explorer.tsx diff --git a/apps/api/src/api/endpoints.ts b/apps/api/src/api/endpoints.ts index fb5b934..cdca601 100644 --- a/apps/api/src/api/endpoints.ts +++ b/apps/api/src/api/endpoints.ts @@ -432,6 +432,16 @@ export const CategoryTrendEndpoint = api.stats.categoryTrend .tags('stats') .operationId('categoryTrend'); +export const CashFlowForecastEndpoint = api.stats.cashFlowForecast + .authorize(PrincipalSchema) + .inject({ db: DbToken, config: ConfigToken, logger: LoggerToken }) + .summary('Cash-flow forecast') + .description( + 'Projects 30-day and 90-day cash flow from historical transactions and detected recurring patterns.' + ) + .tags('stats') + .operationId('cashFlowForecast'); + export const endpoints = { auth: { register: RegisterEndpoint, @@ -503,6 +513,7 @@ export const endpoints = { stats: { overview: StatsOverviewEndpoint, window: StatsWindowEndpoint, - categoryTrend: CategoryTrendEndpoint + categoryTrend: CategoryTrendEndpoint, + cashFlowForecast: CashFlowForecastEndpoint } }; diff --git a/apps/api/src/api/handlers/index.ts b/apps/api/src/api/handlers/index.ts index e3d8e05..06daa36 100644 --- a/apps/api/src/api/handlers/index.ts +++ b/apps/api/src/api/handlers/index.ts @@ -43,6 +43,7 @@ import { transactionScanProgressHandler } from './transaction-scans.js'; import { + cashFlowForecastHandler, categoryTrendHandler, createTransactionHandler, dashboardSummaryHandler, @@ -135,6 +136,7 @@ export const handlers = { stats: { overview: statsOverviewHandler, window: statsWindowHandler, - categoryTrend: categoryTrendHandler + categoryTrend: categoryTrendHandler, + cashFlowForecast: cashFlowForecastHandler } }; diff --git a/apps/api/src/api/handlers/transactions.ts b/apps/api/src/api/handlers/transactions.ts index 95f3a19..2e95d3d 100644 --- a/apps/api/src/api/handlers/transactions.ts +++ b/apps/api/src/api/handlers/transactions.ts @@ -1,4 +1,5 @@ import { ActionResult, type Handler } from '@cleverbrush/server'; +import { cashFlowForecast } from '../../application/cash-flow-forecast.js'; import { categoryTrend, createTransaction, @@ -15,6 +16,7 @@ import { } from '../../application/transactions.js'; import { TransactionCreated } from '../../log-templates.js'; import type { + CashFlowForecastEndpoint, CategoryTrendEndpoint, CreateTransactionEndpoint, DashboardSummaryEndpoint, @@ -162,3 +164,9 @@ export const categoryTrendHandler: Handler< throw err; } }; + +export const cashFlowForecastHandler: Handler< + typeof CashFlowForecastEndpoint +> = async ({ query, principal }, { db, config, logger }) => { + return cashFlowForecast(db, config, logger, principal.userId, query); +}; diff --git a/apps/api/src/application/cash-flow-forecast.test.ts b/apps/api/src/application/cash-flow-forecast.test.ts new file mode 100644 index 0000000..6adebb8 --- /dev/null +++ b/apps/api/src/application/cash-flow-forecast.test.ts @@ -0,0 +1,302 @@ +import { describe, expect, it } from 'vitest'; +import type { + CategoryDb, + TransactionDb, + UserDb, + VendorDb +} from '../db/schemas.js'; +import { buildCashFlowForecast } from './cash-flow-forecast.js'; + +const timestamp = new Date('2026-06-16T12:00:00.000Z'); + +function user(overrides: Partial = {}): UserDb { + return { + id: 1, + authProvider: 'local', + countryCode: 'US', + createdAt: timestamp, + defaultCurrency: 'USD', + email: 'test@example.com', + emailVerified: true, + monthlyEmailReportEnabled: true, + role: 'user', + timezone: 'UTC', + updatedAt: timestamp, + weeklyEmailReportEnabled: true, + ...overrides + }; +} + +function category( + id: number, + name: string, + type: 'expense' | 'income', + overrides: Partial = {} +): CategoryDb { + return { + id, + userId: 1, + archivedAt: null, + createdAt: timestamp, + kind: 'normal', + name, + parentId: null, + type, + updatedAt: timestamp, + ...overrides + }; +} + +function vendor(id: number, name: string): VendorDb { + return { + id, + userId: 1, + createdAt: timestamp, + name, + normalizedName: name.toLowerCase(), + updatedAt: timestamp + }; +} + +function transaction( + id: number, + categoryId: number, + type: 'expense' | 'income', + amount: number, + occurredAt: string, + overrides: Partial = {} +): TransactionDb { + return { + id, + userId: 1, + amount, + categoryId, + createdAt: new Date(occurredAt), + currency: 'USD', + defaultCurrency: 'USD', + defaultCurrencyAmount: amount, + exchangeRate: 1, + exchangeRateDate: occurredAt.slice(0, 10), + occurredAt: new Date(occurredAt), + type, + updatedAt: new Date(occurredAt), + ...overrides + }; +} + +describe('buildCashFlowForecast', () => { + it('returns zero projections when no history is available', () => { + const forecast = buildCashFlowForecast({ + categories: [category(1, 'Groceries', 'expense')], + now: new Date('2026-06-16T08:00:00.000Z'), + transactions: [], + user: user(), + vendors: [] + }); + + expect(forecast.transactionCount).toBe(0); + expect(forecast.recurringPatterns).toEqual([]); + expect( + forecast.windows.map(window => ({ + horizonDays: window.horizonDays, + incomeTotal: window.incomeTotal, + expenseTotal: window.expenseTotal, + confidence: window.confidence + })) + ).toEqual([ + { + horizonDays: 30, + incomeTotal: 0, + expenseTotal: 0, + confidence: 'low' + }, + { + horizonDays: 90, + incomeTotal: 0, + expenseTotal: 0, + confidence: 'low' + } + ]); + }); + + it('projects recurring monthly income and weekly expenses into 30 and 90 day windows', () => { + const salary = category(1, 'Salary', 'income'); + const software = category(2, 'Software', 'expense'); + const payroll = vendor(1, 'Acme Payroll'); + const streaming = vendor(2, 'Streambox'); + const forecast = buildCashFlowForecast({ + categories: [salary, software], + now: new Date('2026-06-16T08:00:00.000Z'), + transactions: [ + transaction( + 1, + salary.id, + 'income', + 3000, + '2026-03-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ), + transaction( + 2, + salary.id, + 'income', + 3000, + '2026-04-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ), + transaction( + 3, + salary.id, + 'income', + 3000, + '2026-05-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ), + transaction( + 4, + salary.id, + 'income', + 3000, + '2026-06-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ), + transaction( + 5, + software.id, + 'expense', + 20, + '2026-05-26T12:00:00.000Z', + { + vendorId: streaming.id + } + ), + transaction( + 6, + software.id, + 'expense', + 20, + '2026-06-02T12:00:00.000Z', + { + vendorId: streaming.id + } + ), + transaction( + 7, + software.id, + 'expense', + 20, + '2026-06-09T12:00:00.000Z', + { + vendorId: streaming.id + } + ) + ], + user: user(), + vendors: [payroll, streaming] + }); + + expect( + forecast.recurringPatterns.map(pattern => ({ + cadence: pattern.cadence, + amount: pattern.amount, + category: pattern.categoryDisplayName, + vendor: pattern.vendorName, + projectedCount: pattern.projectedCount + })) + ).toEqual([ + { + cadence: 'monthly', + amount: 3000, + category: 'Salary', + vendor: 'Acme Payroll', + projectedCount: 3 + }, + { + cadence: 'weekly', + amount: 20, + category: 'Software', + vendor: 'Streambox', + projectedCount: 13 + } + ]); + + const thirty = forecast.windows.find( + window => window.horizonDays === 30 + ); + const ninety = forecast.windows.find( + window => window.horizonDays === 90 + ); + + expect(thirty?.recurringIncomeTotal).toBe(3000); + expect(thirty?.recurringExpenseTotal).toBe(100); + expect(ninety?.recurringIncomeTotal).toBe(9000); + expect(ninety?.recurringExpenseTotal).toBe(260); + }); + + it('does not double-count recurring history in the non-recurring baseline', () => { + const salary = category(1, 'Salary', 'income'); + const payroll = vendor(1, 'Acme Payroll'); + const forecast = buildCashFlowForecast({ + categories: [salary], + now: new Date('2026-06-16T08:00:00.000Z'), + transactions: [ + transaction( + 1, + salary.id, + 'income', + 3000, + '2026-03-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ), + transaction( + 2, + salary.id, + 'income', + 3000, + '2026-04-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ), + transaction( + 3, + salary.id, + 'income', + 3000, + '2026-05-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ), + transaction( + 4, + salary.id, + 'income', + 3000, + '2026-06-01T12:00:00.000Z', + { + vendorId: payroll.id + } + ) + ], + user: user(), + vendors: [payroll] + }); + const thirty = forecast.windows.find( + window => window.horizonDays === 30 + ); + + expect(thirty?.baselineIncomeTotal).toBe(0); + expect(thirty?.incomeTotal).toBe(3000); + }); +}); diff --git a/apps/api/src/application/cash-flow-forecast.ts b/apps/api/src/application/cash-flow-forecast.ts new file mode 100644 index 0000000..aa3d85d --- /dev/null +++ b/apps/api/src/application/cash-flow-forecast.ts @@ -0,0 +1,814 @@ +import type { Logger } from '@cleverbrush/log'; +import type { + CashFlowForecastConfidence, + CashFlowForecastInsight, + CashFlowForecastQuery, + CashFlowForecastResponse, + CashFlowForecastWindow, + CashFlowRecurringPattern +} from '@xpenser/contracts'; +import { + addLocalDays, + addLocalMonths, + dateToLocalDateParam, + formatDateInTimeZone, + localDayDifference, + localEndOfDay, + localStartOfDay +} from '@xpenser/timezone'; +import type { Config } from '../config.js'; +import type { + AppDb, + CategoryDb, + TransactionDb, + UserDb, + VendorDb +} from '../db/schemas.js'; +import { categoryDisplayName, categoryReportingType } from './categories.js'; +import { generateStructuredJson, OpenAIConfigError } from './openai.js'; + +type ForecastType = 'expense' | 'income'; +type ForecastUser = Pick; +type ForecastEvent = { + readonly id: number; + readonly amount: number; + readonly categoryDisplayName: string; + readonly categoryId: number; + readonly occurredAt: Date; + readonly patternKey?: string; + readonly type: ForecastType; + readonly vendorId?: number | null; + readonly vendorName?: string; + readonly note?: string; +}; +type DetectedPattern = CashFlowRecurringPattern & { + readonly intervalDays: number; + readonly lastOccurredAt: Date; + readonly patternKey: string; +}; +type ProjectedOccurrence = { + readonly amount: number; + readonly occurredAt: Date; + readonly patternId: string; + readonly type: ForecastType; +}; + +const forecastHistoryDays = 180; +const forecastHorizons = [30, 90] as const; +const forecastInsightTimeoutMs = 8000; +const maxInsightItems = 4; + +function roundMoney(value: number): number { + return Math.round(value * 100) / 100; +} + +function median(values: readonly number[]): number { + if (values.length === 0) { + return 0; + } + const sorted = [...values].sort((left, right) => left - right); + const middle = Math.floor(sorted.length / 2); + if (sorted.length % 2 === 1) { + return sorted[middle] ?? 0; + } + return ((sorted[middle - 1] ?? 0) + (sorted[middle] ?? 0)) / 2; +} + +function average(values: readonly number[]): number { + if (values.length === 0) { + return 0; + } + return values.reduce((sum, value) => sum + value, 0) / values.length; +} + +function normalizeNote(value?: string | null): string | undefined { + const normalized = value?.trim().toLowerCase().replace(/\s+/g, ' '); + if (!normalized || normalized.length < 3) { + return undefined; + } + return normalized.slice(0, 80); +} + +function forecastPatternKey( + type: ForecastType, + categoryId: number, + vendorId?: number | null, + note?: string +): string | undefined { + if (vendorId) { + return `${type}:category:${categoryId}:vendor:${vendorId}`; + } + if (note) { + return `${type}:category:${categoryId}:note:${note}`; + } + return undefined; +} + +function mapForecastEvent( + row: TransactionDb, + categoriesById: ReadonlyMap, + vendorsById: ReadonlyMap +): ForecastEvent | undefined { + const category = categoriesById.get(row.categoryId) ?? row.category; + const amount = Math.abs(Number(row.defaultCurrencyAmount)); + if (!Number.isFinite(amount) || amount <= 0) { + return undefined; + } + + const type = categoryReportingType(category, row.type); + const vendor = row.vendorId ? vendorsById.get(row.vendorId) : undefined; + const note = normalizeNote(row.note); + const patternKey = forecastPatternKey( + type, + row.categoryId, + row.vendorId, + note + ); + + return { + id: row.id, + amount, + categoryDisplayName: category + ? categoryDisplayName(category, categoriesById) + : '', + categoryId: row.categoryId, + occurredAt: row.occurredAt, + patternKey, + type, + vendorId: row.vendorId ?? null, + vendorName: vendor?.name, + note + }; +} + +function cadenceTarget(cadence: CashFlowRecurringPattern['cadence']): number { + if (cadence === 'weekly') { + return 7; + } + if (cadence === 'biweekly') { + return 14; + } + return 30; +} + +function cadenceTolerance( + cadence: CashFlowRecurringPattern['cadence'] +): number { + return cadence === 'monthly' ? 5 : cadence === 'biweekly' ? 3 : 2; +} + +function cadenceScore( + intervals: readonly number[], + cadence: CashFlowRecurringPattern['cadence'] +) { + const target = cadenceTarget(cadence); + const tolerance = cadenceTolerance(cadence); + const matches = intervals.filter( + interval => Math.abs(interval - target) <= tolerance + ).length; + return { + cadence, + target, + matchRatio: matches / Math.max(1, intervals.length), + averageDelta: average( + intervals.map(interval => Math.abs(interval - target)) + ) + }; +} + +function detectCadence(intervals: readonly number[]) { + const candidates = (['weekly', 'biweekly', 'monthly'] as const) + .map(cadence => cadenceScore(intervals, cadence)) + .sort( + (left, right) => + right.matchRatio - left.matchRatio || + left.averageDelta - right.averageDelta + ); + const best = candidates[0]; + if (!best || best.matchRatio < 0.7) { + return undefined; + } + return best; +} + +function confidence( + occurrenceCount: number, + amountDeviation: number, + cadenceMatchRatio: number +): CashFlowForecastConfidence { + if ( + occurrenceCount >= 4 && + amountDeviation <= 0.1 && + cadenceMatchRatio >= 0.9 + ) { + return 'high'; + } + if ( + occurrenceCount >= 3 && + amountDeviation <= 0.25 && + cadenceMatchRatio >= 0.7 + ) { + return 'medium'; + } + return 'low'; +} + +function nextPatternDate( + value: Date, + cadence: CashFlowRecurringPattern['cadence'], + timeZone: string +): Date { + if (cadence === 'monthly') { + return addLocalMonths(value, 1, timeZone); + } + return addLocalDays(value, cadence === 'biweekly' ? 14 : 7, timeZone); +} + +export function detectRecurringPatterns( + events: readonly ForecastEvent[], + forecastFrom: Date, + forecastTo: Date, + timeZone: string +): DetectedPattern[] { + const groups = new Map(); + for (const event of events) { + if (!event.patternKey) { + continue; + } + groups.set(event.patternKey, [ + ...(groups.get(event.patternKey) ?? []), + event + ]); + } + + const patterns: DetectedPattern[] = []; + for (const [patternKey, group] of groups) { + const sorted = [...group].sort( + (left, right) => + left.occurredAt.getTime() - right.occurredAt.getTime() + ); + if (sorted.length < 3) { + continue; + } + + const intervals = sorted + .slice(1) + .map((event, index) => + localDayDifference( + sorted[index]?.occurredAt ?? event.occurredAt, + event.occurredAt, + timeZone + ) + ); + const cadence = detectCadence(intervals); + if (!cadence) { + continue; + } + + const amounts = sorted.map(event => event.amount); + const amount = median(amounts); + if (amount <= 0) { + continue; + } + + const relativeDeviation = average( + amounts.map(value => Math.abs(value - amount) / amount) + ); + if (relativeDeviation > 0.25) { + continue; + } + + const first = sorted[0]!; + const last = sorted[sorted.length - 1]!; + const daysSinceLast = localDayDifference( + last.occurredAt, + forecastFrom, + timeZone + ); + if (daysSinceLast > cadence.target * 2.5) { + continue; + } + + let nextOccurrence = nextPatternDate( + last.occurredAt, + cadence.cadence, + timeZone + ); + while (nextOccurrence < forecastFrom) { + nextOccurrence = nextPatternDate( + nextOccurrence, + cadence.cadence, + timeZone + ); + } + + const projected = projectPatternOccurrences( + { + amount, + cadence: cadence.cadence, + lastOccurredAt: last.occurredAt, + patternKey + }, + forecastFrom, + addLocalDays(forecastFrom, 90, timeZone), + timeZone + ); + + patterns.push({ + id: patternKey, + amount: roundMoney(amount), + averageIntervalDays: roundMoney(average(intervals)), + cadence: cadence.cadence, + categoryDisplayName: first.categoryDisplayName, + categoryId: first.categoryId, + confidence: confidence( + sorted.length, + relativeDeviation, + cadence.matchRatio + ), + intervalDays: cadence.target, + lastOccurredAt: last.occurredAt, + nextOccurrenceAt: + nextOccurrence <= forecastTo ? nextOccurrence : undefined, + note: first.vendorId ? undefined : first.note, + occurrenceCount: sorted.length, + patternKey, + projectedCount: projected.length, + projectedTotal: roundMoney( + projected.reduce((sum, item) => sum + item.amount, 0) + ), + type: first.type, + vendorId: first.vendorId ?? null, + vendorName: first.vendorName + }); + } + + return patterns.sort( + (left, right) => + Math.abs(right.projectedTotal) - Math.abs(left.projectedTotal) || + left.categoryDisplayName.localeCompare(right.categoryDisplayName) + ); +} + +function projectPatternOccurrences( + pattern: Pick< + DetectedPattern, + 'amount' | 'cadence' | 'lastOccurredAt' | 'patternKey' + >, + from: Date, + toExclusive: Date, + timeZone: string +): ProjectedOccurrence[] { + const occurrences: ProjectedOccurrence[] = []; + let current = nextPatternDate( + pattern.lastOccurredAt, + pattern.cadence, + timeZone + ); + + while (current < from) { + current = nextPatternDate(current, pattern.cadence, timeZone); + } + while (current < toExclusive) { + const [type] = pattern.patternKey.split(':'); + occurrences.push({ + amount: pattern.amount, + occurredAt: current, + patternId: pattern.patternKey, + type: type === 'income' ? 'income' : 'expense' + }); + current = nextPatternDate(current, pattern.cadence, timeZone); + } + + return occurrences; +} + +function windowConfidence( + transactionCount: number, + patterns: readonly DetectedPattern[], + horizonDays: number +): CashFlowForecastConfidence { + if (transactionCount < 10) { + return 'low'; + } + if ( + horizonDays === 30 && + transactionCount >= 45 && + patterns.some(pattern => pattern.confidence === 'high') + ) { + return 'high'; + } + return 'medium'; +} + +function bucketLabel(value: Date, timeZone: string): string { + return formatDateInTimeZone( + value, + timeZone, + { day: 'numeric', month: 'short' }, + 'en-GB' + ); +} + +function buildForecastWindow({ + baselineDailyExpense, + baselineDailyIncome, + forecastFrom, + horizonDays, + patterns, + timeZone, + transactionCount +}: { + readonly baselineDailyExpense: number; + readonly baselineDailyIncome: number; + readonly forecastFrom: Date; + readonly horizonDays: 30 | 90; + readonly patterns: readonly DetectedPattern[]; + readonly timeZone: string; + readonly transactionCount: number; +}): CashFlowForecastWindow { + const toExclusive = addLocalDays(forecastFrom, horizonDays, timeZone); + const to = new Date(toExclusive.getTime() - 1); + const occurrences = patterns.flatMap(pattern => + projectPatternOccurrences(pattern, forecastFrom, toExclusive, timeZone) + ); + const buckets: CashFlowForecastWindow['buckets'] = []; + + for ( + let bucketFrom = forecastFrom; + bucketFrom < toExclusive; + bucketFrom = addLocalDays(bucketFrom, 7, timeZone) + ) { + const bucketToExclusiveCandidate = addLocalDays( + bucketFrom, + 7, + timeZone + ); + const bucketToExclusive = + bucketToExclusiveCandidate < toExclusive + ? bucketToExclusiveCandidate + : toExclusive; + const days = localDayDifference( + bucketFrom, + bucketToExclusive, + timeZone + ); + const bucketOccurrences = occurrences.filter( + occurrence => + occurrence.occurredAt >= bucketFrom && + occurrence.occurredAt < bucketToExclusive + ); + const recurringIncomeTotal = bucketOccurrences + .filter(occurrence => occurrence.type === 'income') + .reduce((sum, occurrence) => sum + occurrence.amount, 0); + const recurringExpenseTotal = bucketOccurrences + .filter(occurrence => occurrence.type === 'expense') + .reduce((sum, occurrence) => sum + occurrence.amount, 0); + const baselineIncomeTotal = baselineDailyIncome * days; + const baselineExpenseTotal = baselineDailyExpense * days; + const incomeTotal = baselineIncomeTotal + recurringIncomeTotal; + const expenseTotal = baselineExpenseTotal + recurringExpenseTotal; + + buckets.push({ + from: bucketFrom, + to: new Date(bucketToExclusive.getTime() - 1), + label: bucketLabel(bucketFrom, timeZone), + baselineExpenseTotal: roundMoney(baselineExpenseTotal), + baselineIncomeTotal: roundMoney(baselineIncomeTotal), + expenseTotal: roundMoney(expenseTotal), + incomeTotal: roundMoney(incomeTotal), + netTotal: roundMoney(incomeTotal - expenseTotal), + recurringExpenseTotal: roundMoney(recurringExpenseTotal), + recurringIncomeTotal: roundMoney(recurringIncomeTotal) + }); + } + + const totals = buckets.reduce( + (sum, bucket) => ({ + baselineExpenseTotal: + sum.baselineExpenseTotal + bucket.baselineExpenseTotal, + baselineIncomeTotal: + sum.baselineIncomeTotal + bucket.baselineIncomeTotal, + expenseTotal: sum.expenseTotal + bucket.expenseTotal, + incomeTotal: sum.incomeTotal + bucket.incomeTotal, + recurringExpenseTotal: + sum.recurringExpenseTotal + bucket.recurringExpenseTotal, + recurringIncomeTotal: + sum.recurringIncomeTotal + bucket.recurringIncomeTotal + }), + { + baselineExpenseTotal: 0, + baselineIncomeTotal: 0, + expenseTotal: 0, + incomeTotal: 0, + recurringExpenseTotal: 0, + recurringIncomeTotal: 0 + } + ); + const netTotal = totals.incomeTotal - totals.expenseTotal; + + return { + horizonDays, + from: forecastFrom, + to, + averageDailyNet: roundMoney(netTotal / horizonDays), + baselineExpenseTotal: roundMoney(totals.baselineExpenseTotal), + baselineIncomeTotal: roundMoney(totals.baselineIncomeTotal), + buckets, + confidence: windowConfidence(transactionCount, patterns, horizonDays), + expenseTotal: roundMoney(totals.expenseTotal), + incomeTotal: roundMoney(totals.incomeTotal), + netTotal: roundMoney(netTotal), + projectedRecurringCount: occurrences.length, + recurringExpenseTotal: roundMoney(totals.recurringExpenseTotal), + recurringIncomeTotal: roundMoney(totals.recurringIncomeTotal) + }; +} + +export function buildCashFlowForecast(input: { + readonly categories: readonly CategoryDb[]; + readonly now: Date; + readonly query?: CashFlowForecastQuery; + readonly transactions: readonly TransactionDb[]; + readonly user: ForecastUser; + readonly vendors: readonly VendorDb[]; +}): CashFlowForecastResponse { + const categoriesById = new Map( + input.categories.map(category => [category.id, category] as const) + ); + const vendorsById = new Map( + input.vendors.map(vendor => [vendor.id, vendor] as const) + ); + const forecastFrom = localStartOfDay( + input.query?.date ?? input.now, + input.user.timezone + ); + const historyFrom = addLocalDays( + forecastFrom, + -forecastHistoryDays, + input.user.timezone + ); + const historyTo = new Date(forecastFrom.getTime() - 1); + const forecastTo = localEndOfDay( + addLocalDays(forecastFrom, 89, input.user.timezone), + input.user.timezone + ); + const events = input.transactions + .map(transaction => + mapForecastEvent(transaction, categoriesById, vendorsById) + ) + .filter((event): event is ForecastEvent => Boolean(event)); + const patterns = detectRecurringPatterns( + events, + forecastFrom, + forecastTo, + input.user.timezone + ); + const recurringPatternKeys = new Set( + patterns.map(pattern => pattern.patternKey) + ); + const nonRecurringEvents = events.filter( + event => + !event.patternKey || !recurringPatternKeys.has(event.patternKey) + ); + const baselineIncomeTotal = nonRecurringEvents + .filter(event => event.type === 'income') + .reduce((sum, event) => sum + event.amount, 0); + const baselineExpenseTotal = nonRecurringEvents + .filter(event => event.type === 'expense') + .reduce((sum, event) => sum + event.amount, 0); + const baselineDailyIncome = baselineIncomeTotal / forecastHistoryDays; + const baselineDailyExpense = baselineExpenseTotal / forecastHistoryDays; + const windows = forecastHorizons.map(horizonDays => + buildForecastWindow({ + baselineDailyExpense, + baselineDailyIncome, + forecastFrom, + horizonDays, + patterns, + timeZone: input.user.timezone, + transactionCount: events.length + }) + ); + + return { + anchorDate: forecastFrom, + currency: input.user.defaultCurrency, + generatedAt: input.now, + historyDays: forecastHistoryDays, + historyFrom, + historyTo, + insightsStatus: 'unavailable', + recurringPatterns: patterns.map( + ({ intervalDays, lastOccurredAt, patternKey, ...pattern }) => + pattern + ), + transactionCount: events.length, + windows + }; +} + +function forecastOpenAiPayload(forecast: CashFlowForecastResponse) { + return { + forecast: { + anchorDate: dateToLocalDateParam(forecast.anchorDate, 'UTC'), + currency: forecast.currency, + dataSemantics: { + totals: 'All totals are deterministic projections in the user default currency.', + baseline: + 'Baseline income and expenses come from non-recurring historical daily averages over the completed history window.', + recurring: + 'Recurring patterns were detected deterministically from repeated category, vendor or note, amount stability, and weekly, biweekly, or monthly cadence.', + limits: 'This is a cash-flow projection from recorded transactions, not a bank balance, investment forecast, or financial advice.' + }, + history: { + from: forecast.historyFrom, + to: forecast.historyTo, + days: forecast.historyDays, + transactions: forecast.transactionCount + }, + windows: forecast.windows.map(window => ({ + horizonDays: window.horizonDays, + incomeTotal: window.incomeTotal, + expenseTotal: window.expenseTotal, + netTotal: window.netTotal, + averageDailyNet: window.averageDailyNet, + confidence: window.confidence, + recurringIncomeTotal: window.recurringIncomeTotal, + recurringExpenseTotal: window.recurringExpenseTotal, + projectedRecurringCount: window.projectedRecurringCount + })), + recurringPatterns: forecast.recurringPatterns.map(pattern => ({ + type: pattern.type, + cadence: pattern.cadence, + amount: pattern.amount, + category: pattern.categoryDisplayName, + vendor: pattern.vendorName, + note: pattern.note, + occurrenceCount: pattern.occurrenceCount, + projectedCount: pattern.projectedCount, + projectedTotal: pattern.projectedTotal, + confidence: pattern.confidence + })) + } + }; +} + +function cleanStringArray(value: unknown): string[] { + return Array.isArray(value) + ? value + .filter((item): item is string => typeof item === 'string') + .map(item => item.trim()) + .filter(Boolean) + .slice(0, maxInsightItems) + : []; +} + +function parseForecastInsight( + value: Partial +): CashFlowForecastInsight { + return { + headline: + typeof value.headline === 'string' && value.headline.trim() + ? value.headline.trim() + : 'Cash-flow forecast ready', + summary: + typeof value.summary === 'string' && value.summary.trim() + ? value.summary.trim() + : 'The forecast is based on recorded transactions and detected recurring patterns.', + risks: cleanStringArray(value.risks), + opportunities: cleanStringArray(value.opportunities), + recurringNotes: cleanStringArray(value.recurringNotes), + actions: cleanStringArray(value.actions) + }; +} + +async function generateForecastInsight( + config: Config, + forecast: CashFlowForecastResponse +): Promise { + const parsed = await generateStructuredJson< + Partial + >(config, { + input: forecastOpenAiPayload(forecast), + model: config.openai.reportModel, + schema: { + additionalProperties: false, + properties: { + headline: { type: 'string' }, + summary: { type: 'string' }, + risks: { + items: { type: 'string' }, + type: 'array' + }, + opportunities: { + items: { type: 'string' }, + type: 'array' + }, + recurringNotes: { + items: { type: 'string' }, + type: 'array' + }, + actions: { + items: { type: 'string' }, + type: 'array' + } + }, + required: [ + 'headline', + 'summary', + 'risks', + 'opportunities', + 'recurringNotes', + 'actions' + ], + type: 'object' + }, + schemaName: 'cash_flow_forecast_insight', + system: [ + 'You write concise personal cash-flow forecast insights.', + 'Use only the provided deterministic projection, recurring pattern context, and confidence fields.', + 'Do not invent balances, vendors, bills, income sources, or transactions.', + 'Do not give investment, credit, tax, or legal advice.', + 'Make clear when confidence is low or when projection depends on detected recurring patterns.' + ].join(' ') + }); + return parseForecastInsight(parsed); +} + +function withTimeout( + promise: Promise, + timeoutMs: number, + message: string +): Promise { + return Promise.race([ + promise, + new Promise((_, reject) => { + setTimeout(() => reject(new Error(message)), timeoutMs); + }) + ]); +} + +export async function cashFlowForecast( + db: AppDb, + config: Config, + logger: Pick, + userId: number, + query: CashFlowForecastQuery +): Promise { + const user = (await db.users.find(userId)) as UserDb | undefined; + if (!user) { + throw new Error('User was not found.'); + } + + const now = new Date(); + const forecastFrom = localStartOfDay(query.date ?? now, user.timezone); + const historyFrom = addLocalDays( + forecastFrom, + -forecastHistoryDays, + user.timezone + ); + const historyTo = new Date(forecastFrom.getTime() - 1); + const [categories, vendors, transactions] = await Promise.all([ + db.categories.where(category => category.userId, userId), + db.vendors.where(vendor => vendor.userId, userId), + db.transactions + .include(transaction => transaction.category) + .where(transaction => transaction.userId, userId) + .where(transaction => transaction.occurredAt, '>=', historyFrom) + .where(transaction => transaction.occurredAt, '<=', historyTo) + ]); + const forecast = buildCashFlowForecast({ + categories: categories as CategoryDb[], + now, + query, + transactions: transactions as TransactionDb[], + user, + vendors: vendors as VendorDb[] + }); + + try { + const insights = await withTimeout( + generateForecastInsight(config, forecast), + forecastInsightTimeoutMs, + 'OpenAI forecast insights timed out.' + ); + return { + ...forecast, + insights, + insightsStatus: 'available' + }; + } catch (err) { + if (err instanceof OpenAIConfigError) { + return forecast; + } + logger.warn('Cash-flow forecast insights failed', { + Error: err instanceof Error ? err.message : String(err), + UserId: userId + }); + return { + ...forecast, + insightsStatus: 'failed' + }; + } +} diff --git a/apps/web/app/(app)/stats/forecast/page.tsx b/apps/web/app/(app)/stats/forecast/page.tsx new file mode 100644 index 0000000..6e3c570 --- /dev/null +++ b/apps/web/app/(app)/stats/forecast/page.tsx @@ -0,0 +1,42 @@ +import Link from 'next/link'; +import { CashFlowForecastExplorer } from '@/components/cash-flow-forecast-explorer'; +import { getApiClient } from '@/lib/api'; +import { parseDateParam } from '@/lib/dashboard-periods'; + +type ForecastSearchParams = { + readonly date?: string; +}; + +export const dynamic = 'force-dynamic'; + +export default async function CashFlowForecastPage({ + searchParams +}: { + readonly searchParams: Promise; +}) { + const params = await searchParams; + const client = await getApiClient({ + retryOnTimeout: false, + timeoutMs: 30_000 + }); + const me = await client.auth.me(); + const date = parseDateParam(params.date, me.timezone); + const forecast = await client.stats.cashFlowForecast({ + query: date ? { date } : {} + }); + + return ( +
+ + Back to reports + + +
+ ); +} diff --git a/apps/web/components/cash-flow-forecast-explorer.test.tsx b/apps/web/components/cash-flow-forecast-explorer.test.tsx new file mode 100644 index 0000000..71fd3dd --- /dev/null +++ b/apps/web/components/cash-flow-forecast-explorer.test.tsx @@ -0,0 +1,139 @@ +/** + * @vitest-environment jsdom + */ + +import { fireEvent, render, screen } from '@testing-library/react'; +import type { CashFlowForecastResponse } from '@xpenser/contracts'; +import type { ReactNode } from 'react'; +import { describe, expect, it, vi } from 'vitest'; +import { CashFlowForecastExplorer } from './cash-flow-forecast-explorer'; + +vi.mock('recharts', () => ({ + Bar: () => null, + BarChart: ({ children }: { readonly children: ReactNode }) => ( +
{children}
+ ), + CartesianGrid: () => null, + ReferenceLine: () => null, + ResponsiveContainer: ({ children }: { readonly children: ReactNode }) => ( +
{children}
+ ), + Tooltip: () => null, + XAxis: () => null, + YAxis: () => null +})); + +function forecast( + overrides: Partial = {} +): CashFlowForecastResponse { + return { + anchorDate: new Date('2026-06-16T00:00:00.000Z'), + currency: 'USD', + generatedAt: new Date('2026-06-16T08:00:00.000Z'), + historyDays: 180, + historyFrom: new Date('2025-12-18T00:00:00.000Z'), + historyTo: new Date('2026-06-15T23:59:59.999Z'), + insightsStatus: 'unavailable', + recurringPatterns: [ + { + id: 'income:category:1:vendor:1', + amount: 3000, + averageIntervalDays: 30.3, + cadence: 'monthly', + categoryDisplayName: 'Salary', + categoryId: 1, + confidence: 'high', + nextOccurrenceAt: new Date('2026-07-01T12:00:00.000Z'), + occurrenceCount: 4, + projectedCount: 3, + projectedTotal: 9000, + type: 'income', + vendorId: 1, + vendorName: 'Acme Payroll' + } + ], + transactionCount: 12, + windows: [ + { + horizonDays: 30, + averageDailyNet: 96.67, + baselineExpenseTotal: 0, + baselineIncomeTotal: 0, + buckets: [], + confidence: 'medium', + expenseTotal: 100, + from: new Date('2026-06-16T00:00:00.000Z'), + incomeTotal: 3000, + netTotal: 2900, + projectedRecurringCount: 5, + recurringExpenseTotal: 100, + recurringIncomeTotal: 3000, + to: new Date('2026-07-15T23:59:59.999Z') + }, + { + horizonDays: 90, + averageDailyNet: 97.11, + baselineExpenseTotal: 0, + baselineIncomeTotal: 0, + buckets: [], + confidence: 'medium', + expenseTotal: 260, + from: new Date('2026-06-16T00:00:00.000Z'), + incomeTotal: 9000, + netTotal: 8740, + projectedRecurringCount: 16, + recurringExpenseTotal: 260, + recurringIncomeTotal: 9000, + to: new Date('2026-09-13T23:59:59.999Z') + } + ], + ...overrides + }; +} + +describe('CashFlowForecastExplorer', () => { + it('renders the 30 day forecast and switches to 90 days', () => { + render( + + ); + + expect( + screen.getByRole('heading', { name: 'Cash-flow forecast' }) + ).toBeTruthy(); + expect(screen.getByText('$2,900.00')).toBeTruthy(); + expect(screen.getByText('AI insight unavailable')).toBeTruthy(); + expect(screen.getByText('Acme Payroll')).toBeTruthy(); + + fireEvent.click(screen.getByRole('button', { name: /90 days/ })); + + expect(screen.getByText('$8,740.00')).toBeTruthy(); + expect( + screen.getByText('16 projected recurring occurrences') + ).toBeTruthy(); + }); + + it('renders available AI insight sections', () => { + render( + + ); + + expect(screen.getByText('Positive cash flow expected')).toBeTruthy(); + expect(screen.getByText('A subscription remains active.')).toBeTruthy(); + expect( + screen.getByText('Review recurring subscriptions.') + ).toBeTruthy(); + }); +}); diff --git a/apps/web/components/cash-flow-forecast-explorer.tsx b/apps/web/components/cash-flow-forecast-explorer.tsx new file mode 100644 index 0000000..c551891 --- /dev/null +++ b/apps/web/components/cash-flow-forecast-explorer.tsx @@ -0,0 +1,468 @@ +'use client'; + +import type { + CashFlowForecastResponse, + CashFlowForecastWindow, + CashFlowRecurringPattern +} from '@xpenser/contracts'; +import { + Badge, + Button, + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, + cn +} from '@xpenser/ui'; +import { + AlertTriangleIcon, + CalendarRangeIcon, + RepeatIcon, + SparklesIcon +} from 'lucide-react'; +import type { ReactNode } from 'react'; +import { useMemo, useState } from 'react'; +import { + Bar, + BarChart, + CartesianGrid, + ReferenceLine, + ResponsiveContainer, + Tooltip, + XAxis, + YAxis +} from 'recharts'; +import { AmountDisplay } from '@/components/amount-display'; +import { formatDate, formatMoney } from '@/lib/format'; + +type Horizon = CashFlowForecastWindow['horizonDays']; +type TooltipPayload = { + readonly color?: string; + readonly name?: string; + readonly value?: number | string; +}; + +const incomeColor = '#047857'; +const expenseColor = '#be123c'; +const netColor = 'hsl(var(--accent))'; + +function forecastWindow( + forecast: CashFlowForecastResponse, + horizon: Horizon +): CashFlowForecastWindow { + return ( + forecast.windows.find(window => window.horizonDays === horizon) ?? + forecast.windows[0]! + ); +} + +function confidenceBadgeClassName(confidence: string): string { + if (confidence === 'high') { + return 'border-emerald-200 bg-emerald-50 text-emerald-700 dark:border-emerald-900 dark:bg-emerald-950/40 dark:text-emerald-300'; + } + if (confidence === 'medium') { + return 'border-amber-200 bg-amber-50 text-amber-700 dark:border-amber-900 dark:bg-amber-950/40 dark:text-amber-300'; + } + return 'border-rose-200 bg-rose-50 text-rose-700 dark:border-rose-900 dark:bg-rose-950/40 dark:text-rose-300'; +} + +function ChartTooltip({ + active, + currency, + label, + payload +}: { + readonly active?: boolean; + readonly currency: string; + readonly label?: string | number; + readonly payload?: readonly TooltipPayload[]; +}) { + if (!active || !payload?.length) { + return null; + } + + return ( +
+

{label}

+
+ {payload.map(item => ( +
+ {item.name} + + {typeof item.value === 'number' + ? formatMoney(item.value, currency) + : item.value} + +
+ ))} +
+
+ ); +} + +function ForecastChart({ + currency, + window +}: { + readonly currency: string; + readonly window: CashFlowForecastWindow; +}) { + const data = window.buckets.map(bucket => ({ + label: bucket.label, + Income: bucket.incomeTotal, + Expenses: -bucket.expenseTotal, + Net: bucket.netTotal + })); + + return ( + + + Weekly forecast + + Baseline averages plus detected recurring patterns. + + + +
+ + + + + + Number(value).toLocaleString('en-US', { + maximumFractionDigits: 0 + }) + } + tickLine={false} + width={48} + /> + + } + /> + + + + + +
+ + + ); +} + +function MetricCard({ + label, + value, + detail +}: { + readonly detail: string; + readonly label: string; + readonly value: ReactNode; +}) { + return ( + + + {label} + {value} + + +

{detail}

+ + + ); +} + +function RecurringPatternRow({ + currency, + pattern +}: { + readonly currency: string; + readonly pattern: CashFlowRecurringPattern; +}) { + return ( +
+
+
+ + + {pattern.vendorName ?? + pattern.note ?? + pattern.categoryDisplayName} + + + {pattern.confidence} + +
+

+ {pattern.categoryDisplayName} - {pattern.cadence},{' '} + {pattern.occurrenceCount} historical occurrences +

+
+
+
+ {formatMoney( + pattern.type === 'expense' + ? -pattern.amount + : pattern.amount, + currency + )} +
+

+ {pattern.projectedCount} next 90 days +

+
+
+ ); +} + +function ForecastInsights({ + forecast +}: { + readonly forecast: CashFlowForecastResponse; +}) { + if (forecast.insightsStatus === 'available' && forecast.insights) { + return ( + + +
+ + {forecast.insights.headline} +
+ + {forecast.insights.summary} + + + + + + + + + + ); + } + + return ( + + +
+ + AI insight unavailable +
+ + The deterministic forecast is still available. AI summaries + require OpenAI configuration and must complete within the + request timeout. + + + + ); +} + +function InsightList({ + items, + title +}: { + readonly items: readonly string[]; + readonly title: string; +}) { + return ( +
+

{title}

+ {items.length > 0 ? ( +
    + {items.map(item => ( +
  • {item}
    • + ))} +
+ ) : ( +

+ No specific items. +

+ )} +
+ ); +} + +export function CashFlowForecastExplorer({ + forecast, + timezone +}: { + readonly forecast: CashFlowForecastResponse; + readonly timezone: string; +}) { + const [horizon, setHorizon] = useState(30); + const selectedWindow = useMemo( + () => forecastWindow(forecast, horizon), + [forecast, horizon] + ); + const recurringPatterns = forecast.recurringPatterns.filter( + pattern => pattern.projectedCount > 0 + ); + + return ( +
+
+
+

+ Cash-flow forecast +

+

+ {formatDate(selectedWindow.from, timezone)} -{' '} + {formatDate(selectedWindow.to, timezone)} in{' '} + {forecast.currency}. +

+
+
+ {([30, 90] as const).map(value => ( + + ))} +
+
+ +
+ + } + /> + + } + /> + + } + /> + + {selectedWindow.confidence} + + } + /> +
+ + + +
+ + + + Recurring patterns + + Detected from repeated category, vendor or note, + amount stability, and cadence. + + + + {recurringPatterns.length > 0 ? ( + recurringPatterns.map(pattern => ( + + )) + ) : ( +

+ No recurring patterns were detected in the + history window. +

+ )} + + +
+
+ ); +} diff --git a/apps/web/components/stats-explorer.tsx b/apps/web/components/stats-explorer.tsx index ac1c52b..ebe9908 100644 --- a/apps/web/components/stats-explorer.tsx +++ b/apps/web/components/stats-explorer.tsx @@ -13,7 +13,11 @@ import { CardHeader, CardTitle } from '@xpenser/ui'; -import { ChevronDownIcon, ChevronRightIcon } from 'lucide-react'; +import { + ChevronDownIcon, + ChevronRightIcon, + TrendingUpIcon +} from 'lucide-react'; import Link from 'next/link'; import { useRouter } from 'next/navigation'; import { @@ -856,17 +860,25 @@ export function StatsExplorer({ return (
-
-

Reports

-

- {formatDashboardRangeLabel({ - from: stats.from, - period: selection.period, - to: stats.to, - timeZone: timezone - })}{' '} - in {stats.currency}. -

+
+
+

Reports

+

+ {formatDashboardRangeLabel({ + from: stats.from, + period: selection.period, + to: stats.to, + timeZone: timezone + })}{' '} + in {stats.currency}. +

+
+
{ api.dashboard.window, api.stats.overview, api.stats.window, - api.stats.categoryTrend + api.stats.categoryTrend, + api.stats.cashFlowForecast ]; for (const endpoint of protectedEndpoints) { diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index b44fb7e..f31a6e0 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -2,6 +2,8 @@ import { array, number } from '@cleverbrush/schema'; import { defineApi, endpoint, route } from '@cleverbrush/server/contract'; import { ApiKeySchema, + CashFlowForecastQuerySchema, + CashFlowForecastResponseSchema, CategoryListQuerySchema, CategorySchema, CategoryTrendQuerySchema, @@ -547,6 +549,17 @@ export const api = defineApi({ .responses({ 200: CategoryTrendResponseSchema, 400: ErrorResponseSchema + }), + cashFlowForecast: endpoint + .get('/api/stats/cash-flow-forecast') + .authorize(PrincipalSchema) + .query(CashFlowForecastQuerySchema) + .cacheTag('stats', request => ({ + date: request.query.date + })) + .responses({ + 200: CashFlowForecastResponseSchema, + 401: ErrorResponseSchema }) } }); diff --git a/packages/contracts/src/schemas.ts b/packages/contracts/src/schemas.ts index 32c9d73..de3d3b2 100644 --- a/packages/contracts/src/schemas.ts +++ b/packages/contracts/src/schemas.ts @@ -1775,6 +1775,246 @@ export const CategoryTrendQuerySchema = object({ }) .schemaName('CategoryTrendQuery'); +export const CashFlowForecastQuerySchema = object({ + /** Date used as the local forecast anchor. Defaults to today. */ + date: date() + .coerce() + .optional() + .describe('Date used as the local forecast anchor. Defaults to today.') +}).schemaName('CashFlowForecastQuery'); + +export const CashFlowForecastConfidenceSchema = enumOf( + 'high', + 'low', + 'medium' +).describe('Forecast confidence based on available history and patterns.'); + +export const CashFlowForecastInsightStatusSchema = enumOf( + 'available', + 'failed', + 'unavailable' +).describe('Whether AI forecast insights were returned.'); + +export const CashFlowRecurringCadenceSchema = enumOf( + 'biweekly', + 'monthly', + 'weekly' +).describe('Detected recurring transaction cadence.'); + +export const CashFlowForecastInsightSchema = object({ + /** AI-generated one-line forecast headline. */ + headline: string().describe('AI-generated one-line forecast headline.'), + /** AI-generated concise forecast summary. */ + summary: string().describe('AI-generated concise forecast summary.'), + /** Forecast risks grounded in the computed projection. */ + risks: array(string()).describe( + 'Forecast risks grounded in the computed projection.' + ), + /** Forecast opportunities grounded in the computed projection. */ + opportunities: array(string()).describe( + 'Forecast opportunities grounded in the computed projection.' + ), + /** Notes about detected recurring patterns. */ + recurringNotes: array(string()).describe( + 'Notes about detected recurring patterns.' + ), + /** Suggested user actions grounded in the computed projection. */ + actions: array(string()).describe( + 'Suggested user actions grounded in the computed projection.' + ) +}).schemaName('CashFlowForecastInsight'); + +export const CashFlowForecastBucketSchema = object({ + /** Bucket start timestamp. */ + from: date().coerce().describe('Bucket start timestamp.'), + /** Bucket end timestamp. */ + to: date().coerce().describe('Bucket end timestamp.'), + /** Short label shown on forecast charts. */ + label: string().describe('Short label shown on forecast charts.'), + /** Projected income in the user's default currency. */ + incomeTotal: decimalNumber().describe( + "Projected income in the user's default currency." + ), + /** Projected expenses in the user's default currency. */ + expenseTotal: decimalNumber().describe( + "Projected expenses in the user's default currency." + ), + /** Projected income minus expenses for the bucket. */ + netTotal: decimalNumber().describe( + 'Projected income minus expenses for the bucket.' + ), + /** Baseline income derived from non-recurring history. */ + baselineIncomeTotal: decimalNumber().describe( + 'Baseline income derived from non-recurring history.' + ), + /** Baseline expenses derived from non-recurring history. */ + baselineExpenseTotal: decimalNumber().describe( + 'Baseline expenses derived from non-recurring history.' + ), + /** Income projected from detected recurring patterns. */ + recurringIncomeTotal: decimalNumber().describe( + 'Income projected from detected recurring patterns.' + ), + /** Expenses projected from detected recurring patterns. */ + recurringExpenseTotal: decimalNumber().describe( + 'Expenses projected from detected recurring patterns.' + ) +}).schemaName('CashFlowForecastBucket'); + +export const CashFlowRecurringPatternSchema = object({ + /** Stable pattern key. */ + id: string().describe('Stable pattern key.'), + /** Detected transaction direction on the reporting side. */ + type: CategoryTypeSchema.describe( + 'Detected transaction direction on the reporting side.' + ), + /** Detected recurring cadence. */ + cadence: CashFlowRecurringCadenceSchema.describe( + 'Detected recurring cadence.' + ), + /** Median transaction amount in the user's default currency. */ + amount: decimalNumber().describe( + "Median transaction amount in the user's default currency." + ), + /** Number of matching historical occurrences. */ + occurrenceCount: number().describe( + 'Number of matching historical occurrences.' + ), + /** Average gap between historical occurrences in local days. */ + averageIntervalDays: decimalNumber().describe( + 'Average gap between historical occurrences in local days.' + ), + /** Category identifier for the pattern. */ + categoryId: number().describe('Category identifier for the pattern.'), + /** Category path for the pattern. */ + categoryDisplayName: string().describe('Category path for the pattern.'), + /** Vendor identifier, when the pattern is vendor-backed. */ + vendorId: number() + .nullable() + .describe('Vendor identifier, when the pattern is vendor-backed.'), + /** Vendor name, when the pattern is vendor-backed. */ + vendorName: string() + .optional() + .describe('Vendor name, when the pattern is vendor-backed.'), + /** Normalized note used when no vendor is linked. */ + note: string() + .optional() + .describe('Normalized note used when no vendor is linked.'), + /** Next projected local occurrence timestamp. */ + nextOccurrenceAt: date() + .coerce() + .optional() + .describe('Next projected local occurrence timestamp.'), + /** Number of projected occurrences in the 90-day forecast window. */ + projectedCount: number().describe( + 'Number of projected occurrences in the 90-day forecast window.' + ), + /** Projected 90-day total for this pattern. */ + projectedTotal: decimalNumber().describe( + 'Projected 90-day total for this pattern.' + ), + /** Pattern confidence based on amount and cadence stability. */ + confidence: CashFlowForecastConfidenceSchema.describe( + 'Pattern confidence based on amount and cadence stability.' + ) +}).schemaName('CashFlowRecurringPattern'); + +export const CashFlowForecastWindowSchema = object({ + /** Forecast horizon in local days. */ + horizonDays: number().describe('Forecast horizon in local days.'), + /** Forecast window start timestamp. */ + from: date().coerce().describe('Forecast window start timestamp.'), + /** Forecast window end timestamp. */ + to: date().coerce().describe('Forecast window end timestamp.'), + /** Projected income in the user's default currency. */ + incomeTotal: decimalNumber().describe( + "Projected income in the user's default currency." + ), + /** Projected expenses in the user's default currency. */ + expenseTotal: decimalNumber().describe( + "Projected expenses in the user's default currency." + ), + /** Projected income minus expenses. */ + netTotal: decimalNumber().describe('Projected income minus expenses.'), + /** Projected average daily net cash flow. */ + averageDailyNet: decimalNumber().describe( + 'Projected average daily net cash flow.' + ), + /** Baseline income derived from non-recurring history. */ + baselineIncomeTotal: decimalNumber().describe( + 'Baseline income derived from non-recurring history.' + ), + /** Baseline expenses derived from non-recurring history. */ + baselineExpenseTotal: decimalNumber().describe( + 'Baseline expenses derived from non-recurring history.' + ), + /** Income projected from detected recurring patterns. */ + recurringIncomeTotal: decimalNumber().describe( + 'Income projected from detected recurring patterns.' + ), + /** Expenses projected from detected recurring patterns. */ + recurringExpenseTotal: decimalNumber().describe( + 'Expenses projected from detected recurring patterns.' + ), + /** Number of recurring occurrences projected into this window. */ + projectedRecurringCount: number().describe( + 'Number of recurring occurrences projected into this window.' + ), + /** Forecast confidence for this horizon. */ + confidence: CashFlowForecastConfidenceSchema.describe( + 'Forecast confidence for this horizon.' + ), + /** Weekly forecast buckets. */ + buckets: array(CashFlowForecastBucketSchema).describe( + 'Weekly forecast buckets.' + ) +}).schemaName('CashFlowForecastWindow'); + +export const CashFlowForecastResponseSchema = object({ + /** Local forecast anchor date. */ + anchorDate: date().coerce().describe('Local forecast anchor date.'), + /** Timestamp when the forecast was generated. */ + generatedAt: date() + .coerce() + .describe('Timestamp when the forecast was generated.'), + /** Currency used for all forecast totals. */ + currency: CurrencyCodeSchema.describe( + 'Currency used for all forecast totals.' + ), + /** Historical range start timestamp used for the forecast. */ + historyFrom: date() + .coerce() + .describe('Historical range start timestamp used for the forecast.'), + /** Historical range end timestamp used for the forecast. */ + historyTo: date() + .coerce() + .describe('Historical range end timestamp used for the forecast.'), + /** Number of historical local days used for baseline projection. */ + historyDays: number().describe( + 'Number of historical local days used for baseline projection.' + ), + /** Number of historical transactions used for projection. */ + transactionCount: number().describe( + 'Number of historical transactions used for projection.' + ), + /** Forecast windows for supported horizons. */ + windows: array(CashFlowForecastWindowSchema).describe( + 'Forecast windows for supported horizons.' + ), + /** Detected recurring transaction patterns. */ + recurringPatterns: array(CashFlowRecurringPatternSchema).describe( + 'Detected recurring transaction patterns.' + ), + /** Whether AI forecast insights were returned. */ + insightsStatus: CashFlowForecastInsightStatusSchema.describe( + 'Whether AI forecast insights were returned.' + ), + /** AI-generated forecast insights, when available. */ + insights: CashFlowForecastInsightSchema.optional().describe( + 'AI-generated forecast insights, when available.' + ) +}).schemaName('CashFlowForecastResponse'); + export const DashboardVendorTotalSchema = object({ /** Vendor identifier, or null for transactions without a vendor. */ vendorId: number() @@ -2347,3 +2587,27 @@ export type CategoryTrendQuery = InferType; export type CategoryTrendResponse = InferType< typeof CategoryTrendResponseSchema >; +export type CashFlowForecastQuery = InferType< + typeof CashFlowForecastQuerySchema +>; +export type CashFlowForecastConfidence = InferType< + typeof CashFlowForecastConfidenceSchema +>; +export type CashFlowForecastInsightStatus = InferType< + typeof CashFlowForecastInsightStatusSchema +>; +export type CashFlowForecastInsight = InferType< + typeof CashFlowForecastInsightSchema +>; +export type CashFlowForecastBucket = InferType< + typeof CashFlowForecastBucketSchema +>; +export type CashFlowRecurringPattern = InferType< + typeof CashFlowRecurringPatternSchema +>; +export type CashFlowForecastWindow = InferType< + typeof CashFlowForecastWindowSchema +>; +export type CashFlowForecastResponse = InferType< + typeof CashFlowForecastResponseSchema +>; diff --git a/tests/e2e/workflows.spec.ts b/tests/e2e/workflows.spec.ts index dbbea2b..5b293c7 100644 --- a/tests/e2e/workflows.spec.ts +++ b/tests/e2e/workflows.spec.ts @@ -115,6 +115,23 @@ test.describe('authenticated app workflows', () => { page.getByRole('heading', { level: 1, name: 'Transactions' }) ).toBeVisible(); + await page.getByRole('link', { name: 'Reports' }).first().click(); + await expect( + page.getByRole('heading', { level: 1, name: 'Reports' }) + ).toBeVisible(); + await page.getByRole('link', { name: 'Forecast' }).click(); + await expect( + page.getByRole('heading', { + level: 1, + name: 'Cash-flow forecast' + }) + ).toBeVisible(); + await expect(page.getByRole('button', { name: /30 days/ })).toBeVisible(); + await page.getByRole('button', { name: /90 days/ }).click(); + await expect( + page.getByText(/projected recurring occurrences/) + ).toBeVisible(); + await page.getByRole('link', { name: 'Add' }).first().click(); await expect( page.getByRole('button', { name: 'Save transaction' }) From 2b44beca63de5a32d828b25c529f9f785d94a240 Mon Sep 17 00:00:00 2001 From: Andrew Zolotukhin Date: Tue, 16 Jun 2026 04:36:06 +0000 Subject: [PATCH 2/5] fix: stabilize forecast chart sizing --- apps/web/components/cash-flow-forecast-explorer.tsx | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/apps/web/components/cash-flow-forecast-explorer.tsx b/apps/web/components/cash-flow-forecast-explorer.tsx index c551891..05c92bc 100644 --- a/apps/web/components/cash-flow-forecast-explorer.tsx +++ b/apps/web/components/cash-flow-forecast-explorer.tsx @@ -127,8 +127,12 @@ function ForecastChart({ -
- +
+ Date: Tue, 16 Jun 2026 04:45:52 +0000 Subject: [PATCH 3/5] fix: keep forecast e2e stable in preview --- apps/api/src/application/cash-flow-forecast.ts | 2 +- tests/e2e/workflows.spec.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/api/src/application/cash-flow-forecast.ts b/apps/api/src/application/cash-flow-forecast.ts index aa3d85d..897cf1e 100644 --- a/apps/api/src/application/cash-flow-forecast.ts +++ b/apps/api/src/application/cash-flow-forecast.ts @@ -55,7 +55,7 @@ type ProjectedOccurrence = { const forecastHistoryDays = 180; const forecastHorizons = [30, 90] as const; -const forecastInsightTimeoutMs = 8000; +const forecastInsightTimeoutMs = 3000; const maxInsightItems = 4; function roundMoney(value: number): number { diff --git a/tests/e2e/workflows.spec.ts b/tests/e2e/workflows.spec.ts index 5b293c7..88af77d 100644 --- a/tests/e2e/workflows.spec.ts +++ b/tests/e2e/workflows.spec.ts @@ -125,7 +125,7 @@ test.describe('authenticated app workflows', () => { level: 1, name: 'Cash-flow forecast' }) - ).toBeVisible(); + ).toBeVisible({ timeout: 15_000 }); await expect(page.getByRole('button', { name: /30 days/ })).toBeVisible(); await page.getByRole('button', { name: /90 days/ }).click(); await expect( From 195033cdb3f0d2449214be71c64b0c8ad497dc01 Mon Sep 17 00:00:00 2001 From: Andrew Zolotukhin Date: Wed, 17 Jun 2026 08:34:01 +0000 Subject: [PATCH 4/5] feat: persist daily cash-flow forecasts --- apps/api/src/api/endpoints.ts | 37 +- apps/api/src/api/handlers/index.ts | 6 +- apps/api/src/api/handlers/transactions.ts | 36 +- .../application/cash-flow-forecast-jobs.ts | 310 ++++++++++++++++ .../cash-flow-forecast-scheduler.ts | 148 ++++++++ .../application/cash-flow-forecast.test.ts | 109 +++++- .../api/src/application/cash-flow-forecast.ts | 346 +++++++++++++++++- apps/api/src/config.ts | 17 + .../db/migrations/018_cash_flow_forecasts.ts | 37 ++ apps/api/src/db/schemas.ts | 36 ++ apps/api/src/index.ts | 8 + .../stats/cash-flow-forecast/jobs/route.ts | 113 ++++++ .../cash-flow-forecast-explorer.test.tsx | 28 ++ .../cash-flow-forecast-explorer.tsx | 284 +++++++++++++- packages/contracts/src/api.test.ts | 4 +- packages/contracts/src/api.ts | 21 +- packages/contracts/src/schemas.ts | 81 ++++ 17 files changed, 1571 insertions(+), 50 deletions(-) create mode 100644 apps/api/src/application/cash-flow-forecast-jobs.ts create mode 100644 apps/api/src/application/cash-flow-forecast-scheduler.ts create mode 100644 apps/api/src/db/migrations/018_cash_flow_forecasts.ts create mode 100644 apps/web/app/api/stats/cash-flow-forecast/jobs/route.ts diff --git a/apps/api/src/api/endpoints.ts b/apps/api/src/api/endpoints.ts index cdca601..85c84b6 100644 --- a/apps/api/src/api/endpoints.ts +++ b/apps/api/src/api/endpoints.ts @@ -434,14 +434,43 @@ export const CategoryTrendEndpoint = api.stats.categoryTrend export const CashFlowForecastEndpoint = api.stats.cashFlowForecast .authorize(PrincipalSchema) - .inject({ db: DbToken, config: ConfigToken, logger: LoggerToken }) + .inject({ + db: DbToken, + knex: KnexToken, + config: ConfigToken, + logger: LoggerToken + }) .summary('Cash-flow forecast') .description( - 'Projects 30-day and 90-day cash flow from historical transactions and detected recurring patterns.' + 'Returns the persisted daily cash-flow forecast, queueing generation when needed.' ) .tags('stats') .operationId('cashFlowForecast'); +export const CashFlowForecastJobEndpoint = api.stats.cashFlowForecastJob + .authorize(PrincipalSchema) + .inject({ + db: DbToken, + knex: KnexToken, + config: ConfigToken, + logger: LoggerToken + }) + .summary('Start cash-flow forecast generation') + .description( + 'Starts an asynchronous forecast generation job and returns a short-lived progress token.' + ) + .tags('stats') + .operationId('startCashFlowForecastJob'); + +export const CashFlowForecastProgressEndpoint = + api.stats.cashFlowForecastProgress + .summary('Cash-flow forecast generation progress') + .description( + 'Streams progress and the final stored forecast for a short-lived forecast job token.' + ) + .tags('stats') + .operationId('cashFlowForecastProgress'); + export const endpoints = { auth: { register: RegisterEndpoint, @@ -514,6 +543,8 @@ export const endpoints = { overview: StatsOverviewEndpoint, window: StatsWindowEndpoint, categoryTrend: CategoryTrendEndpoint, - cashFlowForecast: CashFlowForecastEndpoint + cashFlowForecast: CashFlowForecastEndpoint, + cashFlowForecastJob: CashFlowForecastJobEndpoint, + cashFlowForecastProgress: CashFlowForecastProgressEndpoint } }; diff --git a/apps/api/src/api/handlers/index.ts b/apps/api/src/api/handlers/index.ts index 06daa36..5a14992 100644 --- a/apps/api/src/api/handlers/index.ts +++ b/apps/api/src/api/handlers/index.ts @@ -44,6 +44,7 @@ import { } from './transaction-scans.js'; import { cashFlowForecastHandler, + cashFlowForecastProgressHandler, categoryTrendHandler, createTransactionHandler, dashboardSummaryHandler, @@ -51,6 +52,7 @@ import { deleteTransactionHandler, getTransactionScanImageHandler, listTransactionsHandler, + startCashFlowForecastJobHandler, statsOverviewHandler, statsWindowHandler, updateTransactionHandler @@ -137,6 +139,8 @@ export const handlers = { overview: statsOverviewHandler, window: statsWindowHandler, categoryTrend: categoryTrendHandler, - cashFlowForecast: cashFlowForecastHandler + cashFlowForecast: cashFlowForecastHandler, + cashFlowForecastJob: startCashFlowForecastJobHandler, + cashFlowForecastProgress: cashFlowForecastProgressHandler } }; diff --git a/apps/api/src/api/handlers/transactions.ts b/apps/api/src/api/handlers/transactions.ts index 2e95d3d..ca63b70 100644 --- a/apps/api/src/api/handlers/transactions.ts +++ b/apps/api/src/api/handlers/transactions.ts @@ -1,5 +1,13 @@ -import { ActionResult, type Handler } from '@cleverbrush/server'; +import { + ActionResult, + type Handler, + type SubscriptionHandler +} from '@cleverbrush/server'; import { cashFlowForecast } from '../../application/cash-flow-forecast.js'; +import { + startCashFlowForecastJob, + subscribeCashFlowForecastJob +} from '../../application/cash-flow-forecast-jobs.js'; import { categoryTrend, createTransaction, @@ -17,6 +25,8 @@ import { import { TransactionCreated } from '../../log-templates.js'; import type { CashFlowForecastEndpoint, + CashFlowForecastJobEndpoint, + CashFlowForecastProgressEndpoint, CategoryTrendEndpoint, CreateTransactionEndpoint, DashboardSummaryEndpoint, @@ -167,6 +177,26 @@ export const categoryTrendHandler: Handler< export const cashFlowForecastHandler: Handler< typeof CashFlowForecastEndpoint -> = async ({ query, principal }, { db, config, logger }) => { - return cashFlowForecast(db, config, logger, principal.userId, query); +> = async ({ query, principal }, { db, knex, config, logger }) => { + return cashFlowForecast(db, knex, config, logger, principal.userId, query); +}; + +export const startCashFlowForecastJobHandler: Handler< + typeof CashFlowForecastJobEndpoint +> = async ({ body, principal }, { db, knex, config, logger }) => { + const job = startCashFlowForecastJob( + db, + knex, + config, + logger, + principal.userId, + body + ); + return ActionResult.accepted(job); +}; + +export const cashFlowForecastProgressHandler: SubscriptionHandler< + typeof CashFlowForecastProgressEndpoint +> = async function* ({ query, signal }) { + yield* subscribeCashFlowForecastJob(query, signal); }; diff --git a/apps/api/src/application/cash-flow-forecast-jobs.ts b/apps/api/src/application/cash-flow-forecast-jobs.ts new file mode 100644 index 0000000..56df1d8 --- /dev/null +++ b/apps/api/src/application/cash-flow-forecast-jobs.ts @@ -0,0 +1,310 @@ +import { randomBytes, randomUUID } from 'node:crypto'; +import type { Logger } from '@cleverbrush/log'; +import type { + CashFlowForecastJobBody, + CashFlowForecastJobResponse, + CashFlowForecastProgressEvent, + CashFlowForecastProgressQuery, + CashFlowForecastResponse +} from '@xpenser/contracts'; +import type { Knex } from 'knex'; +import type { Config } from '../config.js'; +import type { AppDb } from '../db/schemas.js'; +import { generateAndPersistCashFlowForecast } from './cash-flow-forecast.js'; + +const jobTtlMs = 30 * 60 * 1_000; + +type MutableForecastJob = { + readonly events: CashFlowForecastProgressEvent[]; + readonly id: string; + readonly listeners: Set<() => void>; + readonly token: string; + readonly userId: number; + deleteTimer: ReturnType | null; + done: boolean; + expiresAt: number; +}; + +type ProgressStage = Exclude< + CashFlowForecastProgressEvent['stage'], + 'complete' | 'failed' | 'queued' +>; + +const jobs = new Map(); + +function scheduleDelete(job: MutableForecastJob): void { + if (job.deleteTimer) { + return; + } + + job.deleteTimer = setTimeout(() => { + if (jobs.get(job.id) === job) { + jobs.delete(job.id); + } + }, jobTtlMs); + job.deleteTimer.unref?.(); +} + +function cleanupExpiredJobs(): void { + const now = Date.now(); + for (const [jobId, job] of jobs) { + if (job.expiresAt <= now) { + jobs.delete(jobId); + if (job.deleteTimer) { + clearTimeout(job.deleteTimer); + } + } + } +} + +function notify(job: MutableForecastJob): void { + for (const listener of job.listeners) { + listener(); + } + job.listeners.clear(); +} + +function event({ + error = null, + forecast = null, + job, + message, + progress, + stage +}: { + readonly error?: string | null; + readonly forecast?: CashFlowForecastResponse | null; + readonly job: MutableForecastJob; + readonly message: string; + readonly progress: number; + readonly stage: CashFlowForecastProgressEvent['stage']; +}): CashFlowForecastProgressEvent { + return { + jobId: job.id, + stage, + message, + progress, + forecast, + error + }; +} + +function emit( + job: MutableForecastJob, + nextEvent: CashFlowForecastProgressEvent +): void { + if (job.done) { + return; + } + + job.events.push(nextEvent); + if (nextEvent.stage === 'complete' || nextEvent.stage === 'failed') { + job.done = true; + job.expiresAt = Date.now() + jobTtlMs; + scheduleDelete(job); + } + notify(job); +} + +function progressMessage(stage: ProgressStage): string { + switch (stage) { + case 'preparing': + return 'Loading transaction history and recurring patterns.'; + case 'analyzing': + return 'Generating AI forecast insight.'; + case 'saving': + return 'Saving daily forecast.'; + } + return 'Generating forecast.'; +} + +function progressValue(stage: ProgressStage): number { + switch (stage) { + case 'preparing': + return 20; + case 'analyzing': + return 55; + case 'saving': + return 90; + } + return 50; +} + +async function runJob( + job: MutableForecastJob, + db: AppDb, + knex: Knex, + config: Config, + logger: Pick, + body: CashFlowForecastJobBody +): Promise { + try { + const result = await generateAndPersistCashFlowForecast( + db, + knex, + config, + logger, + job.userId, + body, + { + force: body.force ?? false, + onProgress: stage => + emit( + job, + event({ + job, + message: progressMessage(stage), + progress: progressValue(stage), + stage + }) + ) + } + ); + if (result.status === 'failed') { + emit( + job, + event({ + error: + result.errorMessage ?? + 'Forecast generation failed. Try again.', + forecast: result.forecast, + job, + message: 'Forecast generation failed.', + progress: 100, + stage: 'failed' + }) + ); + return; + } + + emit( + job, + event({ + forecast: result.forecast, + job, + message: 'Forecast ready.', + progress: 100, + stage: 'complete' + }) + ); + } catch (err) { + emit( + job, + event({ + error: 'Forecast generation failed. Try again.', + job, + message: 'Forecast generation failed.', + progress: 100, + stage: 'failed' + }) + ); + logger.warn('Cash-flow forecast job failed', { + Error: err instanceof Error ? err.message : String(err), + JobId: job.id, + UserId: job.userId + }); + } +} + +function createJob(userId: number): MutableForecastJob { + return { + events: [], + id: randomUUID(), + listeners: new Set(), + token: randomBytes(32).toString('base64url'), + userId, + deleteTimer: null, + done: false, + expiresAt: Date.now() + jobTtlMs + }; +} + +export function startCashFlowForecastJob( + db: AppDb, + knex: Knex, + config: Config, + logger: Pick, + userId: number, + body: CashFlowForecastJobBody +): CashFlowForecastJobResponse { + cleanupExpiredJobs(); + const job = createJob(userId); + jobs.set(job.id, job); + emit( + job, + event({ + job, + message: 'Forecast generation queued.', + progress: 0, + stage: 'queued' + }) + ); + void runJob(job, db, knex, config, logger, body); + return { jobId: job.id, token: job.token }; +} + +function authorizedJob( + query: CashFlowForecastProgressQuery +): MutableForecastJob | undefined { + cleanupExpiredJobs(); + const job = jobs.get(query.jobId); + return job && job.token === query.token ? job : undefined; +} + +function waitForEvent( + job: MutableForecastJob, + signal: AbortSignal +): Promise { + if (signal.aborted || job.done) { + return Promise.resolve(); + } + + return new Promise(resolve => { + let resolved = false; + const listener = () => { + if (resolved) { + return; + } + resolved = true; + job.listeners.delete(listener); + signal.removeEventListener('abort', abortListener); + resolve(); + }; + const abortListener = () => listener(); + job.listeners.add(listener); + signal.addEventListener('abort', abortListener, { once: true }); + }); +} + +export async function* subscribeCashFlowForecastJob( + query: CashFlowForecastProgressQuery, + signal: AbortSignal +): AsyncGenerator { + const job = authorizedJob(query); + if (!job) { + yield { + jobId: query.jobId, + stage: 'failed', + message: 'Forecast job was not found.', + progress: 100, + forecast: null, + error: 'Forecast job was not found.' + }; + return; + } + + let index = 0; + while (!signal.aborted) { + while (index < job.events.length) { + const nextEvent = job.events[index]; + index += 1; + if (nextEvent) { + yield nextEvent; + } + } + if (job.done) { + return; + } + await waitForEvent(job, signal); + } +} diff --git a/apps/api/src/application/cash-flow-forecast-scheduler.ts b/apps/api/src/application/cash-flow-forecast-scheduler.ts new file mode 100644 index 0000000..a44aaaf --- /dev/null +++ b/apps/api/src/application/cash-flow-forecast-scheduler.ts @@ -0,0 +1,148 @@ +import type { Logger } from '@cleverbrush/log'; +import { dateToLocalDateParam } from '@xpenser/timezone'; +import type { Knex } from 'knex'; +import type { Config } from '../config.js'; +import type { AppDb } from '../db/schemas.js'; +import { + cashFlowForecastVersion, + generateAndPersistCashFlowForecast +} from './cash-flow-forecast.js'; + +type SchedulerOptions = { + readonly config: Config; + readonly db: AppDb; + readonly knex: Knex; + readonly logger: Logger; +}; + +type ForecastSchedulerUser = { + readonly id: number; + readonly timezone: string; +}; + +export type CashFlowForecastScheduler = { + readonly stop: () => void; +}; + +async function listForecastUsers(knex: Knex): Promise { + return knex('users as app_user') + .select('app_user.id', 'app_user.timezone') + .whereExists(function activeTransactions() { + this.select(knex.raw('1')) + .from('transactions as txn') + .whereRaw('txn.user_id = app_user.id'); + }); +} + +async function hasForecastForDate( + knex: Knex, + userId: number, + forecastDate: string +): Promise { + const row = await knex('cash_flow_forecasts') + .select('id') + .where({ + user_id: userId, + forecast_date: forecastDate, + forecast_version: cashFlowForecastVersion + }) + .first(); + return Boolean(row); +} + +export async function generateDailyCashFlowForecasts( + db: AppDb, + knex: Knex, + config: Config, + logger: Logger +): Promise { + if (!config.openai.apiKey) { + logger.info( + 'Cash-flow forecast scheduler skipped without OpenAI key', + {} + ); + return 0; + } + + const now = new Date(); + const users = await listForecastUsers(knex); + let generated = 0; + for (const user of users) { + const forecastDate = dateToLocalDateParam(now, user.timezone); + if (await hasForecastForDate(knex, user.id, forecastDate)) { + continue; + } + + const result = await generateAndPersistCashFlowForecast( + db, + knex, + config, + logger, + user.id, + { date: now }, + { force: false } + ); + generated += result.status === 'complete' ? 1 : 0; + } + + return generated; +} + +export function startCashFlowForecastScheduler({ + config, + db, + knex, + logger +}: SchedulerOptions): CashFlowForecastScheduler { + if (!config.cashFlowForecasts.schedulerEnabled || !config.openai.apiKey) { + logger.info('Cash-flow forecast scheduler disabled', {}); + return { stop: () => undefined }; + } + + let running = false; + const run = async () => { + if (running) { + return; + } + running = true; + try { + const generated = await generateDailyCashFlowForecasts( + db, + knex, + config, + logger + ); + logger.info('Daily cash-flow forecasts generated', { + Count: generated + }); + } catch (err) { + logger.error('Cash-flow forecast scheduler failed', { + Error: err instanceof Error ? err.message : String(err) + }); + } finally { + running = false; + } + }; + + const startup = setTimeout(() => { + void run(); + }, 45_000); + startup.unref(); + + const interval = setInterval( + () => { + void run(); + }, + 60 * 60 * 1000 + ); + interval.unref(); + + logger.info('Cash-flow forecast scheduler started', {}); + + return { + stop: () => { + clearTimeout(startup); + clearInterval(interval); + } + }; +} diff --git a/apps/api/src/application/cash-flow-forecast.test.ts b/apps/api/src/application/cash-flow-forecast.test.ts index 6adebb8..c07c10b 100644 --- a/apps/api/src/application/cash-flow-forecast.test.ts +++ b/apps/api/src/application/cash-flow-forecast.test.ts @@ -1,11 +1,17 @@ -import { describe, expect, it } from 'vitest'; +import type { Knex } from 'knex'; +import { describe, expect, it, vi } from 'vitest'; +import type { Config } from '../config.js'; import type { + AppDb, CategoryDb, TransactionDb, UserDb, VendorDb } from '../db/schemas.js'; -import { buildCashFlowForecast } from './cash-flow-forecast.js'; +import { + buildCashFlowForecast, + generateAndPersistCashFlowForecast +} from './cash-flow-forecast.js'; const timestamp = new Date('2026-06-16T12:00:00.000Z'); @@ -84,6 +90,66 @@ function transaction( }; } +function queryRows(rows: readonly T[]) { + let whereCalls = 0; + const query = { + include: () => query, + where: () => { + whereCalls += 1; + return whereCalls >= 3 ? Promise.resolve(rows) : query; + } + }; + return query; +} + +function forecastDb({ + categories, + transactions, + user, + vendors +}: { + readonly categories: readonly CategoryDb[]; + readonly transactions: readonly TransactionDb[]; + readonly user: UserDb; + readonly vendors: readonly VendorDb[]; +}): AppDb { + return { + users: { + find: vi.fn().mockResolvedValue(user) + }, + categories: { + where: vi.fn().mockResolvedValue(categories) + }, + vendors: { + where: vi.fn().mockResolvedValue(vendors) + }, + transactions: queryRows(transactions) + } as unknown as AppDb; +} + +function forecastKnex() { + const inserts: Record[] = []; + const merges: Record[] = []; + const knex = vi.fn(() => ({ + insert: vi.fn((values: Record) => { + inserts.push(values); + return { + onConflict: vi.fn(() => ({ + merge: vi.fn(async (values: Record) => { + merges.push(values); + }) + })) + }; + }) + })); + + return { + inserts, + knex: knex as unknown as Knex, + merges + }; +} + describe('buildCashFlowForecast', () => { it('returns zero projections when no history is available', () => { const forecast = buildCashFlowForecast({ @@ -299,4 +365,43 @@ describe('buildCashFlowForecast', () => { expect(thirty?.baselineIncomeTotal).toBe(0); expect(thirty?.incomeTotal).toBe(3000); }); + + it('persists deterministic forecast when OpenAI is not configured', async () => { + const salary = category(1, 'Salary', 'income'); + const db = forecastDb({ + categories: [salary], + transactions: [], + user: user(), + vendors: [] + }); + const { inserts, knex } = forecastKnex(); + const config = { + openai: { + apiKey: undefined, + reportModel: 'gpt-5-mini' + }, + cashFlowForecasts: { + insightTimeoutMs: 45_000 + } + } as Config; + + const result = await generateAndPersistCashFlowForecast( + db, + knex, + config, + { warn: vi.fn() }, + 1, + { date: new Date('2026-06-16T08:00:00.000Z') }, + { force: true } + ); + + expect(result.status).toBe('complete'); + expect(result.forecast.insightsStatus).toBe('unavailable'); + expect(inserts).toHaveLength(2); + expect(inserts[0]?.status).toBe('pending'); + expect(inserts[1]?.status).toBe('complete'); + expect( + JSON.parse(String(inserts[1]?.forecast_json)).insightsStatus + ).toBe('unavailable'); + }); }); diff --git a/apps/api/src/application/cash-flow-forecast.ts b/apps/api/src/application/cash-flow-forecast.ts index 897cf1e..f25035c 100644 --- a/apps/api/src/application/cash-flow-forecast.ts +++ b/apps/api/src/application/cash-flow-forecast.ts @@ -1,7 +1,9 @@ +import { createHash } from 'node:crypto'; import type { Logger } from '@cleverbrush/log'; import type { CashFlowForecastConfidence, CashFlowForecastInsight, + CashFlowForecastJobBody, CashFlowForecastQuery, CashFlowForecastResponse, CashFlowForecastWindow, @@ -16,6 +18,7 @@ import { localEndOfDay, localStartOfDay } from '@xpenser/timezone'; +import type { Knex } from 'knex'; import type { Config } from '../config.js'; import type { AppDb, @@ -28,6 +31,7 @@ import { categoryDisplayName, categoryReportingType } from './categories.js'; import { generateStructuredJson, OpenAIConfigError } from './openai.js'; type ForecastType = 'expense' | 'income'; +type ForecastLogger = Pick; type ForecastUser = Pick; type ForecastEvent = { readonly id: number; @@ -55,9 +59,46 @@ type ProjectedOccurrence = { const forecastHistoryDays = 180; const forecastHorizons = [30, 90] as const; -const forecastInsightTimeoutMs = 3000; +export const cashFlowForecastVersion = '1'; const maxInsightItems = 4; +type StoredCashFlowForecastStatus = 'complete' | 'failed' | 'pending'; +type StoredCashFlowForecastRow = { + readonly id: number; + readonly user_id: number; + readonly forecast_date: string; + readonly forecast_version: string; + readonly input_hash: string; + readonly model: string; + readonly status: StoredCashFlowForecastStatus; + readonly forecast_json: string; + readonly error_message?: string | null; + readonly generated_at?: Date | string | null; + readonly created_at: Date | string; + readonly updated_at: Date | string; +}; +type ForecastInput = { + readonly categories: readonly CategoryDb[]; + readonly forecast: CashFlowForecastResponse; + readonly forecastDate: string; + readonly inputHash: string; + readonly transactions: readonly TransactionDb[]; + readonly user: UserDb; + readonly vendors: readonly VendorDb[]; +}; +type ForecastGenerationResult = { + readonly errorMessage?: string; + readonly forecast: CashFlowForecastResponse; + readonly status: StoredCashFlowForecastStatus; +}; +type ForecastGenerationStage = 'preparing' | 'analyzing' | 'saving'; +type ForecastGenerationOptions = { + readonly force?: boolean; + readonly onProgress?: (stage: ForecastGenerationStage) => void; +}; + +const autoGenerationKeys = new Set(); + function roundMoney(value: number): number { return Math.round(value * 100) / 100; } @@ -654,6 +695,112 @@ function forecastOpenAiPayload(forecast: CashFlowForecastResponse) { }; } +function cashFlowForecastInputHash(forecast: CashFlowForecastResponse): string { + return createHash('sha256') + .update(JSON.stringify(forecastOpenAiPayload(forecast))) + .digest('hex'); +} + +function coerceDate(value: Date | string | undefined): Date | undefined { + return value ? new Date(value) : undefined; +} + +function coerceForecastDates( + value: CashFlowForecastResponse +): CashFlowForecastResponse { + return { + ...value, + anchorDate: coerceDate(value.anchorDate)!, + generatedAt: coerceDate(value.generatedAt)!, + historyFrom: coerceDate(value.historyFrom)!, + historyTo: coerceDate(value.historyTo)!, + recurringPatterns: value.recurringPatterns.map(pattern => ({ + ...pattern, + nextOccurrenceAt: coerceDate(pattern.nextOccurrenceAt) + })), + windows: value.windows.map(window => ({ + ...window, + from: coerceDate(window.from)!, + to: coerceDate(window.to)!, + buckets: window.buckets.map(bucket => ({ + ...bucket, + from: coerceDate(bucket.from)!, + to: coerceDate(bucket.to)! + })) + })) + }; +} + +function parseStoredForecast( + row: StoredCashFlowForecastRow +): CashFlowForecastResponse | undefined { + try { + return coerceForecastDates(JSON.parse(row.forecast_json)); + } catch { + return undefined; + } +} + +async function readStoredForecast( + knex: Knex, + userId: number, + forecastDate: string +): Promise { + const rows = await knex('cash_flow_forecasts') + .where({ + user_id: userId, + forecast_date: forecastDate, + forecast_version: cashFlowForecastVersion + }) + .limit(1); + return rows[0]; +} + +async function upsertStoredForecast( + knex: Knex, + values: { + readonly errorMessage?: string | null; + readonly forecast: CashFlowForecastResponse; + readonly forecastDate: string; + readonly inputHash: string; + readonly model: string; + readonly status: StoredCashFlowForecastStatus; + readonly userId: number; + } +): Promise { + const now = new Date(); + await knex('cash_flow_forecasts') + .insert({ + user_id: values.userId, + forecast_date: values.forecastDate, + forecast_version: cashFlowForecastVersion, + input_hash: values.inputHash, + model: values.model, + status: values.status, + forecast_json: JSON.stringify(values.forecast), + error_message: values.errorMessage ?? null, + generated_at: + values.status === 'complete' || values.status === 'failed' + ? values.forecast.generatedAt + : null, + created_at: now, + updated_at: now + }) + .onConflict(['user_id', 'forecast_date', 'forecast_version']) + .merge({ + input_hash: values.inputHash, + model: values.model, + status: values.status, + forecast_json: JSON.stringify(values.forecast), + error_message: values.errorMessage ?? null, + generated_at: + values.status === 'complete' || values.status === 'failed' + ? values.forecast.generatedAt + : null, + updated_at: now + }); +} + function cleanStringArray(value: unknown): string[] { return Array.isArray(value) ? value @@ -741,21 +888,24 @@ function withTimeout( timeoutMs: number, message: string ): Promise { + let timeout: ReturnType | undefined; return Promise.race([ promise, new Promise((_, reject) => { - setTimeout(() => reject(new Error(message)), timeoutMs); + timeout = setTimeout(() => reject(new Error(message)), timeoutMs); }) - ]); + ]).finally(() => { + if (timeout) { + clearTimeout(timeout); + } + }); } -export async function cashFlowForecast( +async function loadForecastInput( db: AppDb, - config: Config, - logger: Pick, userId: number, - query: CashFlowForecastQuery -): Promise { + query: CashFlowForecastQuery = {} +): Promise { const user = (await db.users.find(userId)) as UserDb | undefined; if (!user) { throw new Error('User was not found.'); @@ -763,6 +913,7 @@ export async function cashFlowForecast( const now = new Date(); const forecastFrom = localStartOfDay(query.date ?? now, user.timezone); + const forecastDate = dateToLocalDateParam(forecastFrom, user.timezone); const historyFrom = addLocalDays( forecastFrom, -forecastHistoryDays, @@ -786,29 +937,184 @@ export async function cashFlowForecast( user, vendors: vendors as VendorDb[] }); + const inputHash = cashFlowForecastInputHash(forecast); + + return { + categories: categories as CategoryDb[], + forecast, + forecastDate, + inputHash, + transactions: transactions as TransactionDb[], + user, + vendors: vendors as VendorDb[] + }; +} + +function failedForecastMessage(err: unknown): string { + if (err instanceof Error && err.message.includes('timed out')) { + return 'AI forecast insight timed out. Try regenerating later.'; + } + return 'AI forecast insight unavailable. Try regenerating later.'; +} + +export async function generateAndPersistCashFlowForecast( + db: AppDb, + knex: Knex, + config: Config, + logger: ForecastLogger, + userId: number, + body: CashFlowForecastJobBody = {}, + options: ForecastGenerationOptions = {} +): Promise { + options.onProgress?.('preparing'); + const input = await loadForecastInput(db, userId, body); + const stored = options.force + ? undefined + : await readStoredForecast(knex, userId, input.forecastDate); + if ( + stored && + stored.input_hash === input.inputHash && + (stored.status === 'complete' || stored.status === 'failed') + ) { + const forecast = parseStoredForecast(stored); + if (forecast) { + return { + errorMessage: stored.error_message ?? undefined, + forecast, + status: stored.status + }; + } + } + + await upsertStoredForecast(knex, { + forecast: { + ...input.forecast, + insightsStatus: 'pending' + }, + forecastDate: input.forecastDate, + inputHash: input.inputHash, + model: config.openai.reportModel, + status: 'pending', + userId + }); + + let errorMessage: string | undefined; + let finalForecast: CashFlowForecastResponse; + let status: StoredCashFlowForecastStatus = 'complete'; try { + options.onProgress?.('analyzing'); const insights = await withTimeout( - generateForecastInsight(config, forecast), - forecastInsightTimeoutMs, + generateForecastInsight(config, input.forecast), + config.cashFlowForecasts.insightTimeoutMs, 'OpenAI forecast insights timed out.' ); - return { - ...forecast, + finalForecast = { + ...input.forecast, + generatedAt: new Date(), insights, insightsStatus: 'available' }; } catch (err) { if (err instanceof OpenAIConfigError) { - return forecast; + finalForecast = { + ...input.forecast, + generatedAt: new Date(), + insightsStatus: 'unavailable' + }; + } else { + errorMessage = failedForecastMessage(err); + status = 'failed'; + logger.warn('Cash-flow forecast insights failed', { + Error: err instanceof Error ? err.message : String(err), + ForecastDate: input.forecastDate, + UserId: userId + }); + finalForecast = { + ...input.forecast, + generatedAt: new Date(), + insightsStatus: 'failed' + }; } - logger.warn('Cash-flow forecast insights failed', { - Error: err instanceof Error ? err.message : String(err), - UserId: userId + } + + options.onProgress?.('saving'); + await upsertStoredForecast(knex, { + errorMessage, + forecast: finalForecast, + forecastDate: input.forecastDate, + inputHash: input.inputHash, + model: config.openai.reportModel, + status, + userId + }); + + return { + errorMessage, + forecast: finalForecast, + status + }; +} + +function scheduleMissingForecastGeneration( + db: AppDb, + knex: Knex, + config: Config, + logger: ForecastLogger, + userId: number, + input: ForecastInput +): void { + if (!config.openai.apiKey) { + return; + } + + const key = `${userId}:${input.forecastDate}:${input.inputHash}`; + if (autoGenerationKeys.has(key)) { + return; + } + + autoGenerationKeys.add(key); + void generateAndPersistCashFlowForecast( + db, + knex, + config, + logger, + userId, + { date: input.forecast.anchorDate }, + { force: false } + ) + .catch(err => { + logger.warn('Cash-flow forecast background generation failed', { + Error: err instanceof Error ? err.message : String(err), + ForecastDate: input.forecastDate, + UserId: userId + }); + }) + .finally(() => { + autoGenerationKeys.delete(key); }); - return { - ...forecast, - insightsStatus: 'failed' - }; +} + +export async function cashFlowForecast( + db: AppDb, + knex: Knex, + config: Config, + logger: ForecastLogger, + userId: number, + query: CashFlowForecastQuery +): Promise { + const input = await loadForecastInput(db, userId, query); + const stored = await readStoredForecast(knex, userId, input.forecastDate); + if (stored && stored.input_hash === input.inputHash) { + const forecast = parseStoredForecast(stored); + if (forecast) { + return forecast; + } } + + scheduleMissingForecastGeneration(db, knex, config, logger, userId, input); + return { + ...input.forecast, + insightsStatus: config.openai.apiKey ? 'pending' : 'unavailable' + }; } diff --git a/apps/api/src/config.ts b/apps/api/src/config.ts index 62e16e8..8326278 100644 --- a/apps/api/src/config.ts +++ b/apps/api/src/config.ts @@ -125,6 +125,16 @@ export const config = parseEnv( number().coerce().default(3) ) }, + cashFlowForecastsEnv: { + schedulerEnabled: env( + 'CASH_FLOW_FORECAST_SCHEDULER_ENABLED', + string().default('1') + ), + insightTimeoutMs: env( + 'CASH_FLOW_FORECAST_INSIGHT_TIMEOUT_MS', + number().coerce().default(45_000) + ) + }, logLevel: env( 'LOG_LEVEL', string() @@ -146,6 +156,9 @@ export const config = parseEnv( const schedulerEnabled = ['1', 'true', 'yes'].includes( base.emailReportsEnv.schedulerEnabled.toLowerCase() ); + const cashFlowForecastSchedulerEnabled = ['1', 'true', 'yes'].includes( + base.cashFlowForecastsEnv.schedulerEnabled.toLowerCase() + ); const passport = applyHostedPassportDefaults( base.app.url, base.passport @@ -167,6 +180,10 @@ export const config = parseEnv( deliveryHourLocal: base.emailReportsEnv.deliveryHourLocal, maxAttempts: base.emailReportsEnv.maxAttempts }, + cashFlowForecasts: { + schedulerEnabled: cashFlowForecastSchedulerEnabled, + insightTimeoutMs: base.cashFlowForecastsEnv.insightTimeoutMs + }, vendorEnrichment: { enabled: ['1', 'true', 'yes'].includes( diff --git a/apps/api/src/db/migrations/018_cash_flow_forecasts.ts b/apps/api/src/db/migrations/018_cash_flow_forecasts.ts new file mode 100644 index 0000000..21a1668 --- /dev/null +++ b/apps/api/src/db/migrations/018_cash_flow_forecasts.ts @@ -0,0 +1,37 @@ +import type { Knex } from 'knex'; + +export async function up(knex: Knex): Promise { + await knex.schema.createTable('cash_flow_forecasts', table => { + table.increments('id').primary(); + table + .integer('user_id') + .notNullable() + .references('id') + .inTable('users') + .onDelete('CASCADE'); + table.date('forecast_date').notNullable(); + table.string('forecast_version', 32).notNullable(); + table.string('input_hash', 128).notNullable(); + table.string('model', 128).notNullable(); + table.string('status', 32).notNullable(); + table.text('forecast_json').notNullable(); + table.text('error_message'); + table.timestamp('generated_at'); + table.timestamp('created_at').notNullable().defaultTo(knex.fn.now()); + table.timestamp('updated_at').notNullable().defaultTo(knex.fn.now()); + + table.unique( + ['user_id', 'forecast_date', 'forecast_version'], + 'uq_cash_flow_forecasts_user_date_version' + ); + table.index( + ['user_id', 'forecast_date'], + 'idx_cash_flow_forecasts_user_date' + ); + table.index(['status'], 'idx_cash_flow_forecasts_status'); + }); +} + +export async function down(knex: Knex): Promise { + await knex.schema.dropTableIfExists('cash_flow_forecasts'); +} diff --git a/apps/api/src/db/schemas.ts b/apps/api/src/db/schemas.ts index 41d02af..7486d24 100644 --- a/apps/api/src/db/schemas.ts +++ b/apps/api/src/db/schemas.ts @@ -347,6 +347,25 @@ export const TransactionScanImageDbSchema = object({ updatedAt: date().hasColumnName('updated_at').defaultTo('now') }).hasTableName('transaction_scan_images'); +export const CashFlowForecastDbSchema = object({ + id: number().primaryKey(), + userId: number() + .hasColumnName('user_id') + .references('users', 'id') + .onDelete('CASCADE') + .index('idx_cash_flow_forecasts_user_id'), + forecastDate: string().hasColumnName('forecast_date'), + forecastVersion: string().hasColumnName('forecast_version'), + inputHash: string().hasColumnName('input_hash'), + model: string(), + status: string(), + forecastJson: string().hasColumnName('forecast_json'), + errorMessage: string().hasColumnName('error_message').optional(), + generatedAt: date().hasColumnName('generated_at').optional(), + createdAt: date().hasColumnName('created_at').defaultTo('now'), + updatedAt: date().hasColumnName('updated_at').defaultTo('now') +}).hasTableName('cash_flow_forecasts'); + export const ExchangeRateDbSchema = object({ id: number().primaryKey(), baseCurrency: string().hasColumnName('base_currency'), @@ -398,6 +417,7 @@ export const TransactionScanItemEntity = defineEntity( export const TransactionScanImageEntity = defineEntity( TransactionScanImageDbSchema ); +export const CashFlowForecastEntity = defineEntity(CashFlowForecastDbSchema); export const ExchangeRateEntity = defineEntity(ExchangeRateDbSchema); export const entityMap = { @@ -417,6 +437,7 @@ export const entityMap = { transactionScans: TransactionScanEntity, transactionScanItems: TransactionScanItemEntity, transactionScanImages: TransactionScanImageEntity, + cashFlowForecasts: CashFlowForecastEntity, exchangeRates: ExchangeRateEntity }; @@ -612,3 +633,18 @@ export type TransactionScanImageDb = { readonly createdAt: Date; readonly updatedAt: Date; }; + +export type CashFlowForecastDb = { + readonly id: number; + readonly userId: number; + readonly forecastDate: string; + readonly forecastVersion: string; + readonly inputHash: string; + readonly model: string; + readonly status: string; + readonly forecastJson: string; + readonly errorMessage?: string | null; + readonly generatedAt?: Date | null; + readonly createdAt: Date; + readonly updatedAt: Date; +}; diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index ba01daa..58303a5 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -6,6 +6,7 @@ import { } from '@cleverbrush/log'; import { otelLogSink, traceEnricher } from '@cleverbrush/otel'; import knex from 'knex'; +import { startCashFlowForecastScheduler } from './application/cash-flow-forecast-scheduler.js'; import { startEmailReportScheduler } from './application/email-report-scheduler.js'; import { config } from './config.js'; import { runMigrations } from './db/migrate.js'; @@ -43,6 +44,12 @@ async function main() { knex: dbResources.knex, logger }); + const cashFlowForecastScheduler = startCashFlowForecastScheduler({ + config, + db: dbResources.db, + knex: dbResources.knex, + logger + }); const server = buildServer(config, logger, dbResources); const httpServer = await server.listen(config.api.port, config.api.host); logger.info(ApiListening, { @@ -56,6 +63,7 @@ async function main() { await httpServer.close(); } finally { emailReportScheduler.stop(); + cashFlowForecastScheduler.stop(); await dbResources.knex.destroy(); await logger.dispose(); await otel.shutdown(); diff --git a/apps/web/app/api/stats/cash-flow-forecast/jobs/route.ts b/apps/web/app/api/stats/cash-flow-forecast/jobs/route.ts new file mode 100644 index 0000000..36a2751 --- /dev/null +++ b/apps/web/app/api/stats/cash-flow-forecast/jobs/route.ts @@ -0,0 +1,113 @@ +import { createXpenserClient } from '@xpenser/client'; +import type { CashFlowForecastJobResponse } from '@xpenser/contracts'; +import { NextResponse } from 'next/server'; +import { auth } from '@/auth'; +import { webConfig } from '@/lib/config'; + +export const dynamic = 'force-dynamic'; + +type ForecastJobRouteResponse = + | { readonly error: string; readonly job?: undefined } + | { + readonly error?: undefined; + readonly job: CashFlowForecastJobResponse; + }; + +type ForecastJobRequestBody = { + readonly date?: string; + readonly force?: boolean; +}; + +function apiErrorStatus(err: unknown): number | undefined { + return typeof err === 'object' && + err !== null && + 'status' in err && + typeof err.status === 'number' + ? err.status + : undefined; +} + +function apiErrorMessage(err: unknown): string | undefined { + const body = + typeof err === 'object' && err !== null && 'body' in err + ? (err as { readonly body?: unknown }).body + : undefined; + return typeof body === 'object' && + body !== null && + 'message' in body && + typeof body.message === 'string' + ? body.message + : undefined; +} + +function errorResponse(message: string, status: number) { + return NextResponse.json( + { error: message }, + { status } + ); +} + +function requestBody(value: unknown): ForecastJobRequestBody | undefined { + if (typeof value !== 'object' || value === null) { + return {}; + } + + const record = value as Record; + if (record.date !== undefined && typeof record.date !== 'string') { + return undefined; + } + if (record.force !== undefined && typeof record.force !== 'boolean') { + return undefined; + } + + return { + date: record.date, + force: record.force + }; +} + +export async function POST(request: Request) { + const session = await auth(); + if (!session?.apiToken) { + return errorResponse('Session expired.', 401); + } + + let body: ForecastJobRequestBody | undefined; + try { + body = requestBody(await request.json()); + } catch { + body = {}; + } + + if (!body) { + return errorResponse('Invalid forecast request.', 400); + } + + const client = createXpenserClient({ + baseUrl: webConfig.apiBaseUrl, + getToken: () => session.apiToken, + retryOnTimeout: false + }); + + try { + const job = await client.stats.cashFlowForecastJob({ + body: { + date: body.date ? new Date(body.date) : undefined, + force: body.force ?? false + } + }); + return NextResponse.json({ job }); + } catch (err) { + const status = apiErrorStatus(err); + if (status === 400) { + return errorResponse( + apiErrorMessage(err) ?? 'Could not start forecast generation.', + 400 + ); + } + if (status === 401) { + return errorResponse('Session expired.', 401); + } + return errorResponse('Could not start forecast generation.', 500); + } +} diff --git a/apps/web/components/cash-flow-forecast-explorer.test.tsx b/apps/web/components/cash-flow-forecast-explorer.test.tsx index 71fd3dd..10dcc5b 100644 --- a/apps/web/components/cash-flow-forecast-explorer.test.tsx +++ b/apps/web/components/cash-flow-forecast-explorer.test.tsx @@ -8,6 +8,19 @@ import type { ReactNode } from 'react'; import { describe, expect, it, vi } from 'vitest'; import { CashFlowForecastExplorer } from './cash-flow-forecast-explorer'; +const mocks = vi.hoisted(() => ({ + createXpenserClient: vi.fn(), + refresh: vi.fn() +})); + +vi.mock('@xpenser/client', () => ({ + createXpenserClient: mocks.createXpenserClient +})); + +vi.mock('next/navigation', () => ({ + useRouter: () => ({ refresh: mocks.refresh }) +})); + vi.mock('recharts', () => ({ Bar: () => null, BarChart: ({ children }: { readonly children: ReactNode }) => ( @@ -101,6 +114,8 @@ describe('CashFlowForecastExplorer', () => { screen.getByRole('heading', { name: 'Cash-flow forecast' }) ).toBeTruthy(); expect(screen.getByText('$2,900.00')).toBeTruthy(); + expect(screen.getByRole('button', { name: 'Regenerate' })).toBeTruthy(); + expect(screen.getByText(/Generated/)).toBeTruthy(); expect(screen.getByText('AI insight unavailable')).toBeTruthy(); expect(screen.getByText('Acme Payroll')).toBeTruthy(); @@ -136,4 +151,17 @@ describe('CashFlowForecastExplorer', () => { screen.getByText('Review recurring subscriptions.') ).toBeTruthy(); }); + + it('renders pending AI insight progress', () => { + render( + + ); + + expect(screen.getByText('AI insight generating')).toBeTruthy(); + expect(screen.getByRole('status')).toBeTruthy(); + expect(screen.getByText('20%')).toBeTruthy(); + }); }); diff --git a/apps/web/components/cash-flow-forecast-explorer.tsx b/apps/web/components/cash-flow-forecast-explorer.tsx index 05c92bc..12125ab 100644 --- a/apps/web/components/cash-flow-forecast-explorer.tsx +++ b/apps/web/components/cash-flow-forecast-explorer.tsx @@ -1,6 +1,9 @@ 'use client'; +import { createXpenserClient } from '@xpenser/client'; import type { + CashFlowForecastJobResponse, + CashFlowForecastProgressEvent, CashFlowForecastResponse, CashFlowForecastWindow, CashFlowRecurringPattern @@ -18,9 +21,13 @@ import { import { AlertTriangleIcon, CalendarRangeIcon, + ClockIcon, + LoaderCircleIcon, + RefreshCwIcon, RepeatIcon, SparklesIcon } from 'lucide-react'; +import { useRouter } from 'next/navigation'; import type { ReactNode } from 'react'; import { useMemo, useState } from 'react'; import { @@ -34,7 +41,7 @@ import { YAxis } from 'recharts'; import { AmountDisplay } from '@/components/amount-display'; -import { formatDate, formatMoney } from '@/lib/format'; +import { formatDate, formatDateTime, formatMoney } from '@/lib/format'; type Horizon = CashFlowForecastWindow['horizonDays']; type TooltipPayload = { @@ -42,11 +49,77 @@ type TooltipPayload = { readonly name?: string; readonly value?: number | string; }; +type ForecastJobRouteResponse = + | { readonly error: string; readonly job?: undefined } + | { + readonly error?: undefined; + readonly job: CashFlowForecastJobResponse; + }; +type RegenerationState = { + readonly error?: string; + readonly message: string; + readonly progress: number; + readonly running: boolean; +}; const incomeColor = '#047857'; const expenseColor = '#be123c'; const netColor = 'hsl(var(--accent))'; +function browserApiBaseUrl(): string { + const configured = process.env.NEXT_PUBLIC_API_BASE_URL; + if (configured) { + return configured.replace(/\/$/, ''); + } + if ( + window.location.hostname === 'localhost' && + window.location.port === '3000' + ) { + return 'http://localhost:4000'; + } + return new URL('/external-api', window.location.href) + .toString() + .replace(/\/$/, ''); +} + +function forecastError(event: CashFlowForecastProgressEvent): string { + return event.error ?? 'Forecast generation failed. Try again.'; +} + +async function waitForForecastJob( + job: CashFlowForecastJobResponse, + onProgress: (update: RegenerationState) => void +): Promise { + const client = createXpenserClient({ + baseUrl: browserApiBaseUrl(), + retryOnTimeout: false + }); + const subscription = client.stats.cashFlowForecastProgress({ + query: { jobId: job.jobId, token: job.token }, + reconnect: { maxRetries: 3, backoffLimit: 5_000 } + }); + + try { + for await (const event of subscription) { + onProgress({ + message: event.message, + progress: Math.min(100, Math.max(0, event.progress)), + running: event.stage !== 'complete' && event.stage !== 'failed' + }); + if (event.stage === 'failed') { + throw new Error(forecastError(event)); + } + if (event.stage === 'complete') { + return; + } + } + } finally { + subscription.close(); + } + + throw new Error('Could not connect to forecast progress. Try again.'); +} + function forecastWindow( forecast: CashFlowForecastResponse, horizon: Horizon @@ -286,6 +359,40 @@ function ForecastInsights({ ); } + if (forecast.insightsStatus === 'pending') { + return ( + + +
+ + AI insight generating +
+ + The deterministic forecast is ready while the daily AI + summary is being generated in the background. + + + + ); + } + + if (forecast.insightsStatus === 'failed') { + return ( + + +
+ + AI insight failed +
+ + The deterministic forecast is still available. Use + regenerate to retry the AI summary. + + + + ); + } + return ( @@ -295,8 +402,7 @@ function ForecastInsights({
The deterministic forecast is still available. AI summaries - require OpenAI configuration and must complete within the - request timeout. + require OpenAI configuration. @@ -335,7 +441,16 @@ export function CashFlowForecastExplorer({ readonly forecast: CashFlowForecastResponse; readonly timezone: string; }) { + const router = useRouter(); const [horizon, setHorizon] = useState(30); + const [regeneration, setRegeneration] = useState({ + message: + forecast.insightsStatus === 'pending' + ? 'Forecast generation is running.' + : '', + progress: forecast.insightsStatus === 'pending' ? 20 : 0, + running: false + }); const selectedWindow = useMemo( () => forecastWindow(forecast, horizon), [forecast, horizon] @@ -343,6 +458,62 @@ export function CashFlowForecastExplorer({ const recurringPatterns = forecast.recurringPatterns.filter( pattern => pattern.projectedCount > 0 ); + const regenerating = regeneration.running; + const showProgress = + regenerating || + Boolean(regeneration.error) || + forecast.insightsStatus === 'pending'; + + async function regenerateForecast(): Promise { + setRegeneration({ + message: 'Starting forecast regeneration.', + progress: 0, + running: true + }); + + try { + const response = await fetch('/api/stats/cash-flow-forecast/jobs', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + date: + forecast.anchorDate instanceof Date + ? forecast.anchorDate.toISOString() + : forecast.anchorDate, + force: true + }) + }); + const result = (await response + .json() + .catch(() => null)) as ForecastJobRouteResponse | null; + if (!response.ok || !result || result.error || !result.job) { + throw new Error( + result?.error ?? 'Could not start forecast generation.' + ); + } + + await waitForForecastJob(result.job, update => { + setRegeneration(update); + }); + setRegeneration({ + message: 'Forecast ready.', + progress: 100, + running: false + }); + router.refresh(); + } catch (err) { + setRegeneration({ + error: + err instanceof Error + ? err.message + : 'Forecast generation failed. Try again.', + message: 'Forecast generation failed.', + progress: 100, + running: false + }); + router.refresh(); + } + } return (
@@ -356,28 +527,103 @@ export function CashFlowForecastExplorer({ {formatDate(selectedWindow.to, timezone)} in{' '} {forecast.currency}.

+

+ + Generated{' '} + {formatDateTime(forecast.generatedAt, timezone)} +

-
- {([30, 90] as const).map(value => ( - - ))} + /> + Regenerate + +
+ {([30, 90] as const).map(value => ( + + ))} +
+ {showProgress ? ( +
+
+
+ {regeneration.error ? ( + + ) : ( + + )} + + {regeneration.error || + regeneration.message || + 'Forecast generation is running.'} + +
+ + {Math.round(regeneration.progress)}% + +
+ {!regeneration.error ? ( +
+
+
+ ) : null} +
+ ) : null} +
{ expect(authRoles(api.telegram.link)).toBeNull(); expect(authRoles(api.telegram.token)).toBeNull(); expect(authRoles(api.transactionScans.progress)).toBeNull(); + expect(authRoles(api.stats.cashFlowForecastProgress)).toBeNull(); }); it('marks user-scoped endpoints as authenticated', () => { @@ -62,7 +63,8 @@ describe('api contract authorization metadata', () => { api.stats.overview, api.stats.window, api.stats.categoryTrend, - api.stats.cashFlowForecast + api.stats.cashFlowForecast, + api.stats.cashFlowForecastJob ]; for (const endpoint of protectedEndpoints) { diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index f31a6e0..63f2196 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -2,6 +2,10 @@ import { array, number } from '@cleverbrush/schema'; import { defineApi, endpoint, route } from '@cleverbrush/server/contract'; import { ApiKeySchema, + CashFlowForecastJobBodySchema, + CashFlowForecastJobResponseSchema, + CashFlowForecastProgressEventSchema, + CashFlowForecastProgressQuerySchema, CashFlowForecastQuerySchema, CashFlowForecastResponseSchema, CategoryListQuerySchema, @@ -77,6 +81,7 @@ const CategoryMoveAndDelete = route({ id: number().coerce() })`/${t => const VendorEnrich = route({ id: number().coerce() })`/${t => t.id}/enrich`; const StatsCategoryTrend = route({ id: number().coerce() })`/categories/${t => t.id}/trend`; +const CashFlowForecastJobs = route`/cash-flow-forecast/jobs`; const TransactionScanDecision = route({ scanId: number().coerce(), itemId: number().coerce() @@ -560,7 +565,21 @@ export const api = defineApi({ .responses({ 200: CashFlowForecastResponseSchema, 401: ErrorResponseSchema - }) + }), + cashFlowForecastJob: stats + .post(CashFlowForecastJobs) + .body(CashFlowForecastJobBodySchema) + .clearsCacheTag('stats') + .responses({ + 202: CashFlowForecastJobResponseSchema, + 400: ErrorResponseSchema, + 401: ErrorResponseSchema + }), + cashFlowForecastProgress: endpoint + .subscription('/api/stats/cash-flow-forecast/jobs/progress') + .public() + .query(CashFlowForecastProgressQuerySchema) + .outgoing(CashFlowForecastProgressEventSchema) } }); diff --git a/packages/contracts/src/schemas.ts b/packages/contracts/src/schemas.ts index de3d3b2..51e0297 100644 --- a/packages/contracts/src/schemas.ts +++ b/packages/contracts/src/schemas.ts @@ -1783,6 +1783,20 @@ export const CashFlowForecastQuerySchema = object({ .describe('Date used as the local forecast anchor. Defaults to today.') }).schemaName('CashFlowForecastQuery'); +export const CashFlowForecastJobBodySchema = object({ + /** Date used as the local forecast anchor. Defaults to today. */ + date: date() + .coerce() + .optional() + .describe('Date used as the local forecast anchor. Defaults to today.'), + /** Force regeneration even when a cached forecast already exists. */ + force: boolean() + .optional() + .describe( + 'Force regeneration even when a cached forecast already exists.' + ) +}).schemaName('CashFlowForecastJobBody'); + export const CashFlowForecastConfidenceSchema = enumOf( 'high', 'low', @@ -1792,6 +1806,7 @@ export const CashFlowForecastConfidenceSchema = enumOf( export const CashFlowForecastInsightStatusSchema = enumOf( 'available', 'failed', + 'pending', 'unavailable' ).describe('Whether AI forecast insights were returned.'); @@ -2015,6 +2030,60 @@ export const CashFlowForecastResponseSchema = object({ ) }).schemaName('CashFlowForecastResponse'); +export const CashFlowForecastJobResponseSchema = object({ + /** Short-lived forecast job identifier used by the progress subscription. */ + jobId: string().describe( + 'Short-lived forecast job identifier used by the progress subscription.' + ), + /** One-time token scoped to this forecast job. */ + token: string().describe('One-time token scoped to this forecast job.') +}).schemaName('CashFlowForecastJobResponse'); + +export const CashFlowForecastProgressQuerySchema = object({ + /** Short-lived forecast job identifier returned by the start endpoint. */ + jobId: string() + .required('forecast job is required') + .nonempty('forecast job is required') + .describe( + 'Short-lived forecast job identifier returned by the start endpoint.' + ), + /** One-time token scoped to this forecast job. */ + token: string() + .required('forecast token is required') + .nonempty('forecast token is required') + .describe('One-time token scoped to this forecast job.') +}).schemaName('CashFlowForecastProgressQuery'); + +export const CashFlowForecastProgressStageSchema = enumOf( + 'queued', + 'preparing', + 'analyzing', + 'saving', + 'complete', + 'failed' +).describe('Current forecast job stage.'); + +export const CashFlowForecastProgressEventSchema = object({ + /** Short-lived forecast job identifier. */ + jobId: string().describe('Short-lived forecast job identifier.'), + /** Current forecast job stage. */ + stage: CashFlowForecastProgressStageSchema.describe( + 'Current forecast job stage.' + ), + /** User-facing progress message. */ + message: string().describe('User-facing progress message.'), + /** Approximate forecast progress from 0 to 100. */ + progress: number().describe('Approximate forecast progress from 0 to 100.'), + /** Final forecast result when the job completed successfully. */ + forecast: union(CashFlowForecastResponseSchema) + .or(nul()) + .describe('Final forecast result when the job completed successfully.'), + /** Safe user-facing failure message when the job failed. */ + error: string() + .nullable() + .describe('Safe user-facing failure message when the job failed.') +}).schemaName('CashFlowForecastProgressEvent'); + export const DashboardVendorTotalSchema = object({ /** Vendor identifier, or null for transactions without a vendor. */ vendorId: number() @@ -2590,6 +2659,9 @@ export type CategoryTrendResponse = InferType< export type CashFlowForecastQuery = InferType< typeof CashFlowForecastQuerySchema >; +export type CashFlowForecastJobBody = InferType< + typeof CashFlowForecastJobBodySchema +>; export type CashFlowForecastConfidence = InferType< typeof CashFlowForecastConfidenceSchema >; @@ -2611,3 +2683,12 @@ export type CashFlowForecastWindow = InferType< export type CashFlowForecastResponse = InferType< typeof CashFlowForecastResponseSchema >; +export type CashFlowForecastJobResponse = InferType< + typeof CashFlowForecastJobResponseSchema +>; +export type CashFlowForecastProgressQuery = InferType< + typeof CashFlowForecastProgressQuerySchema +>; +export type CashFlowForecastProgressEvent = InferType< + typeof CashFlowForecastProgressEventSchema +>; From 511d45a9aa02ee707e5dcbc167cce3941f692cda Mon Sep 17 00:00:00 2001 From: Andrew Zolotukhin Date: Wed, 17 Jun 2026 08:54:39 +0000 Subject: [PATCH 5/5] fix: poll persisted forecast regeneration --- .../app/api/stats/cash-flow-forecast/route.ts | 41 ++++++ .../cash-flow-forecast-explorer.test.tsx | 5 - .../cash-flow-forecast-explorer.tsx | 120 ++++++++++-------- 3 files changed, 110 insertions(+), 56 deletions(-) create mode 100644 apps/web/app/api/stats/cash-flow-forecast/route.ts diff --git a/apps/web/app/api/stats/cash-flow-forecast/route.ts b/apps/web/app/api/stats/cash-flow-forecast/route.ts new file mode 100644 index 0000000..f74fa3e --- /dev/null +++ b/apps/web/app/api/stats/cash-flow-forecast/route.ts @@ -0,0 +1,41 @@ +import type { CashFlowForecastResponse } from '@xpenser/contracts'; +import { NextResponse } from 'next/server'; +import { getApiClient } from '@/lib/api'; + +export const dynamic = 'force-dynamic'; + +type ForecastRouteResponse = + | { readonly error: string; readonly forecast?: undefined } + | { + readonly error?: undefined; + readonly forecast: CashFlowForecastResponse; + }; + +function errorResponse(message: string, status: number) { + return NextResponse.json( + { error: message }, + { status } + ); +} + +export async function GET(request: Request) { + const url = new URL(request.url); + const date = url.searchParams.get('date'); + const parsedDate = date ? new Date(date) : undefined; + if (date && Number.isNaN(parsedDate?.getTime())) { + return errorResponse('Invalid forecast date.', 400); + } + + try { + const client = await getApiClient({ + retryOnTimeout: false, + timeoutMs: 30_000 + }); + const forecast = await client.stats.cashFlowForecast({ + query: parsedDate ? { date: parsedDate } : {} + }); + return NextResponse.json({ forecast }); + } catch { + return errorResponse('Could not load forecast.', 500); + } +} diff --git a/apps/web/components/cash-flow-forecast-explorer.test.tsx b/apps/web/components/cash-flow-forecast-explorer.test.tsx index 10dcc5b..44ed7b3 100644 --- a/apps/web/components/cash-flow-forecast-explorer.test.tsx +++ b/apps/web/components/cash-flow-forecast-explorer.test.tsx @@ -9,14 +9,9 @@ import { describe, expect, it, vi } from 'vitest'; import { CashFlowForecastExplorer } from './cash-flow-forecast-explorer'; const mocks = vi.hoisted(() => ({ - createXpenserClient: vi.fn(), refresh: vi.fn() })); -vi.mock('@xpenser/client', () => ({ - createXpenserClient: mocks.createXpenserClient -})); - vi.mock('next/navigation', () => ({ useRouter: () => ({ refresh: mocks.refresh }) })); diff --git a/apps/web/components/cash-flow-forecast-explorer.tsx b/apps/web/components/cash-flow-forecast-explorer.tsx index 12125ab..04f3155 100644 --- a/apps/web/components/cash-flow-forecast-explorer.tsx +++ b/apps/web/components/cash-flow-forecast-explorer.tsx @@ -1,9 +1,7 @@ 'use client'; -import { createXpenserClient } from '@xpenser/client'; import type { CashFlowForecastJobResponse, - CashFlowForecastProgressEvent, CashFlowForecastResponse, CashFlowForecastWindow, CashFlowRecurringPattern @@ -55,6 +53,12 @@ type ForecastJobRouteResponse = readonly error?: undefined; readonly job: CashFlowForecastJobResponse; }; +type ForecastRouteResponse = + | { readonly error: string; readonly forecast?: undefined } + | { + readonly error?: undefined; + readonly forecast: CashFlowForecastResponse; + }; type RegenerationState = { readonly error?: string; readonly message: string; @@ -66,58 +70,65 @@ const incomeColor = '#047857'; const expenseColor = '#be123c'; const netColor = 'hsl(var(--accent))'; -function browserApiBaseUrl(): string { - const configured = process.env.NEXT_PUBLIC_API_BASE_URL; - if (configured) { - return configured.replace(/\/$/, ''); - } - if ( - window.location.hostname === 'localhost' && - window.location.port === '3000' - ) { - return 'http://localhost:4000'; - } - return new URL('/external-api', window.location.href) - .toString() - .replace(/\/$/, ''); +function delay(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); } -function forecastError(event: CashFlowForecastProgressEvent): string { - return event.error ?? 'Forecast generation failed. Try again.'; +function forecastAnchorDate(value: Date | string): string { + return value instanceof Date ? value.toISOString() : value; } -async function waitForForecastJob( - job: CashFlowForecastJobResponse, - onProgress: (update: RegenerationState) => void -): Promise { - const client = createXpenserClient({ - baseUrl: browserApiBaseUrl(), - retryOnTimeout: false - }); - const subscription = client.stats.cashFlowForecastProgress({ - query: { jobId: job.jobId, token: job.token }, - reconnect: { maxRetries: 3, backoffLimit: 5_000 } +async function fetchStoredForecast( + anchorDate: Date | string +): Promise { + const params = new URLSearchParams({ + date: forecastAnchorDate(anchorDate) }); + const response = await fetch(`/api/stats/cash-flow-forecast?${params}`); + const result = (await response + .json() + .catch(() => null)) as ForecastRouteResponse | null; + if (!response.ok || !result || result.error || !result.forecast) { + throw new Error(result?.error ?? 'Could not load forecast.'); + } + return result.forecast; +} - try { - for await (const event of subscription) { - onProgress({ - message: event.message, - progress: Math.min(100, Math.max(0, event.progress)), - running: event.stage !== 'complete' && event.stage !== 'failed' - }); - if (event.stage === 'failed') { - throw new Error(forecastError(event)); - } - if (event.stage === 'complete') { - return; - } +async function waitForStoredForecast({ + anchorDate, + onProgress, + previousGeneratedAt +}: { + readonly anchorDate: Date | string; + readonly onProgress: (update: RegenerationState) => void; + readonly previousGeneratedAt: Date | string; +}): Promise { + const previousTime = new Date(previousGeneratedAt).getTime(); + const startedAt = Date.now(); + while (Date.now() - startedAt < 90_000) { + await delay(2_000); + const forecast = await fetchStoredForecast(anchorDate); + const generatedAt = new Date(forecast.generatedAt).getTime(); + if ( + forecast.insightsStatus !== 'pending' && + generatedAt > previousTime + ) { + return; } - } finally { - subscription.close(); + onProgress({ + message: + forecast.insightsStatus === 'pending' + ? 'Generating AI forecast insight.' + : 'Waiting for stored forecast.', + progress: Math.min( + 90, + Math.round(35 + ((Date.now() - startedAt) / 90_000) * 55) + ), + running: true + }); } - throw new Error('Could not connect to forecast progress. Try again.'); + throw new Error('Forecast generation is still running. Try again shortly.'); } function forecastWindow( @@ -472,14 +483,12 @@ export function CashFlowForecastExplorer({ }); try { + const anchorDate = forecastAnchorDate(forecast.anchorDate); const response = await fetch('/api/stats/cash-flow-forecast/jobs', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ - date: - forecast.anchorDate instanceof Date - ? forecast.anchorDate.toISOString() - : forecast.anchorDate, + date: anchorDate, force: true }) }); @@ -492,8 +501,17 @@ export function CashFlowForecastExplorer({ ); } - await waitForForecastJob(result.job, update => { - setRegeneration(update); + setRegeneration({ + message: 'Forecast generation queued.', + progress: 15, + running: true + }); + await waitForStoredForecast({ + anchorDate, + previousGeneratedAt: forecast.generatedAt, + onProgress: update => { + setRegeneration(update); + } }); setRegeneration({ message: 'Forecast ready.',