import { BadRequestException, Injectable } from '@nestjs/common'; import { add, type Decimal, decimalFromString, formatDecimal, isZero, } from '../../../common/decimal'; import { MoneyDto } from '../../../common/money.dto'; import { ISO_DATE_LENGTH, ISO_DATE_START } from '../../../common/time'; import { InstrumentMetadataService, metadataKey, TbankAccessService, } from '../../../integrations/tbank/tbank.api'; import { CatalogEnrichmentService } from '../instruments/instruments.api'; import { investmentResourceMeta } from '../investment-resource.mapper'; import { ForecastService, type PortfolioPositionDto, PortfolioService, } from '../portfolio/portfolio.api'; import { hourlyAttempt, TbankSyncRepository, } from '../tbank-sync/tbank-sync.api'; import { bondEvent, dividendEvent } from './upcoming-event.mapper'; import { UpcomingEventDto, UpcomingEventsDto, UpcomingEventsQueryDto, } from './upcoming-events.dto'; export enum EventsErrorCode { InvalidPeriod = 'INVALID_EVENTS_PERIOD' } const FIRST_BATCH_OFFSET = 0; const DEFAULT_EVENT_WINDOW_DAYS = 30; const calendarReadBatchSize = 100; type Position = PortfolioPositionDto; @Injectable() export class UpcomingEventsService { constructor( private readonly access: TbankAccessService, private readonly portfolios: PortfolioService, private readonly sources: CatalogEnrichmentService, private readonly forecasts: ForecastService, private readonly metadata: InstrumentMetadataService, private readonly sync: TbankSyncRepository, ) {} // Read endpoints stay local; callers without a saved portfolio receive unknown. async get( userId: string, accountId: string, query: UpcomingEventsQueryDto = {}, ): Promise>> { const window = dateWindow(query); const connection = await this.access.local(userId, accountId); const portfolio = await this.portfolios.read(userId, accountId); const hourly = await this.sync.hourly(userId); const events: UpcomingEventDto[] = []; let ready = portfolio.data !== null; const positions = portfolio.data?.positions.filter(supported) ?? []; const { sources, metadata } = await this.readSources( userId, positions, window, true, ); for (const position of positions) { const generic = metadata.get(metadataKey(position))?.data; const uid = position.instrumentUid || generic?.uid; if (!uid) { ready = false; sources.push({ input: { uid: '', instrumentType: 'share', kind: 'detail' }, version: null, payload: undefined, instrument: null, price: null, meta: investmentResourceMeta({ resourceKey: `events/identity/${position.figi ?? ''}`, dataRevision: '0', availability: 'missing', fetchedAt: null, freshForMs: 0, work: null, }), }); continue; } const type = position.type === 'STOCK' ? 'share' : 'bond'; const calendar = sources.find( (source) => source.input.uid === uid && source.input.kind === (type === 'share' ? 'dividends' : 'bond-events'), ); const instrument = sources.find( (source) => source.input.uid === uid && source.input.kind === 'detail', )?.instrument || generic; const ticker = position.ticker || instrument?.ticker; if (!calendar || calendar.meta.completeness !== 'complete' || !ticker) { ready = false; continue; } const identity = { figi: position.figi || null, instrumentUid: uid, ticker, displayName: instrument?.name?.trim() || position.displayName || ticker, isin: instrument?.isin || position.isin || null, logoName: instrument?.brand?.logoName || position.logoName || null, }; const quantity = decimalFromString(position.quantity); const projected = type === 'share' ? (calendar.payload?.dividends ?? []) .filter( (event) => event.dividendType?.toLowerCase() !== 'cancelled' && Boolean(event.recordDate), ) .map((event) => dividendEvent(identity, quantity, event)) : (calendar.payload?.bondEvents ?? []).map((event) => bondEvent(identity, quantity, event), ); events.push( ...projected .filter( (event) => event.date >= window.from && event.date <= window.to, ) .map((event) => ({ ...event, ...(accountId === 'all' && position.account ? { account: position.account } : {}), })), ); } events.sort( (a, b) => a.date.localeCompare(b.date) || (a.account?.id ?? '').localeCompare(b.account?.id ?? '') || a.type.localeCompare(b.type) || a.ticker.localeCompare(b.ticker), ); const candidate: Omit | undefined = ready ? { accountId, calculatedAt: new Date().toISOString(), events, incomeTotalsByCurrency: incomeTotals(events), } : undefined; return this.forecasts.result({ userId, revision: connection.revision, membershipRevision: connection.membershipRevision, scopeKey: `events/${JSON.stringify([accountId, window.from, window.to])}`, replacesLatest: query.from === undefined && query.to === undefined, portfolio: portfolio.meta, lastAttempt: hourlyAttempt( hourly, 'events', accountId === 'all' ? undefined : accountId, ), sources, candidate, }); } getAll(userId: string, query: UpcomingEventsQueryDto = {}) { return this.get(userId, 'all', query); } // Triggers online refreshes of the account's event sources and returns the // launched work ids for the hourly walk to observe. async refresh( userId: string, accountId: string, query: UpcomingEventsQueryDto = {}, ): Promise { const window = dateWindow(query); await this.access.local(userId, accountId); const portfolio = await this.portfolios.read(userId, accountId); const positions = portfolio.data?.positions.filter(supported) ?? []; const { launched } = await this.readSources( userId, positions, window, false, ); return launched; } private async readSources( userId: string, positions: Position[], window: { from: string; to: string }, observe: boolean, ) { const metadata = await this.metadata.savedMany(positions); const sources: Awaited> = []; const launched: string[] = []; for (const type of ['share', 'bond'] as const) { const identities = this.positionIdentities(positions, metadata, type); const uids = [...new Set(identities.map((entry) => entry.uid))]; for ( let offset = FIRST_BATCH_OFFSET; offset < uids.length; offset += calendarReadBatchSize ) { const batch = await this.sources.readMany( userId, type, uids.slice(offset, offset + calendarReadBatchSize), observe, ['detail', type === 'share' ? 'dividends' : 'bond-events'], { ...window, tickers: Object.fromEntries( identities.map((entry) => [entry.uid, entry.ticker]), ), }, ); sources.push(...batch); for (const source of batch) { if (source.meta.sync.state === 'running' && source.meta.sync.id) { launched.push(source.meta.sync.id); } } } } return { sources, metadata, launched }; } private positionIdentities( positions: Position[], metadata: Awaited>, type: 'bond' | 'share', ) { return positions .filter((position) => position.type === (type === 'share' ? 'STOCK' : 'BOND')) .map((position) => ({ uid: position.instrumentUid || metadata.get(metadataKey(position))?.data?.uid, ticker: position.ticker || metadata.get(metadataKey(position))?.data?.ticker || '', })) .filter((entry): entry is { ticker: string; uid: string } => Boolean(entry.uid), ); } } function supported(position: Position): boolean { return ( !isZero(decimalFromString(position.quantity)) && ['BOND', 'STOCK'].includes(position.type) ); } function dateWindow(query: UpcomingEventsQueryDto) { const from = query.from ?? new Date().toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH); const end = new Date(`${from}T00:00:00.000Z`); end.setUTCDate(end.getUTCDate() + DEFAULT_EVENT_WINDOW_DAYS); const to = query.to ?? end.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH); if (from > to) { throw new BadRequestException({ code: EventsErrorCode.InvalidPeriod }); } return { from, to }; } function incomeTotals(events: UpcomingEventDto[]): MoneyDto[] { const totals = new Map(); for (const event of events) { if (!['coupon', 'dividend'].includes(event.type) || !event.projectedAmount) { continue; } const current = totals.get(event.projectedAmount.currency) ?? { value: 0n, scale: 0, }; totals.set( event.projectedAmount.currency, add(current, decimalFromString(event.projectedAmount.value)), ); } return [...totals] .sort(([left], [right]) => left.localeCompare(right)) .map(([currency, value]) => ({ currency, value: formatDecimal(value) })); }