feat: add hourly portfolio sync history

src/modules/investments/upcoming-events/upcoming-events.service.ts

@@ -45,64 +45,24 @@ export class UpcomingEventsService {
     private readonly metadata: InstrumentMetadataService,
   ) {}
 
+  // Read endpoints stay local; callers without a saved portfolio receive unknown.
   async get(
     userId: string,
     accountId: string,
     query: UpcomingEventsQueryDto = {},
-  ) {
+  ): Promise<Awaited<ReturnType<ForecastService['result']>>> {
     const window = dateWindow(query);
     const connection = await this.access.local(userId, accountId);
-    const portfolio = await this.portfolios.read(
-      userId,
-      accountId,
-      query.sync === 'observe',
-    );
-    const sources: Awaited<ReturnType<CatalogEnrichmentService['readMany']>> =
-      [];
+    const portfolio = await this.portfolios.read(userId, accountId);
     const events: UpcomingEventDto[] = [];
     let ready = portfolio.data !== null;
     const positions = portfolio.data?.positions.filter(supported) ?? [];
-    const metadata = await this.metadata.savedMany(positions);
-    for (const type of ['share', 'bond'] as const) {
-      const identities = 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),
-        );
-      const uids = [...new Set(identities.map((entry) => entry.uid))];
-      for (
-        let offset = FIRST_BATCH_OFFSET;
-        offset < uids.length;
-        offset += calendarReadBatchSize
-      ) {
-        sources.push(
-          ...(await this.sources.readMany(
-            userId,
-            type,
-            uids.slice(offset, offset + calendarReadBatchSize),
-            query.sync === 'observe',
-            ['detail', type === 'share' ? 'dividends' : 'bond-events'],
-            {
-              ...window,
-              tickers: Object.fromEntries(
-                identities.map((entry) => [entry.uid, entry.ticker]),
-              ),
-            },
-          )),
-        );
-      }
-    }
+    const { sources, metadata } = await this.readSources(
+      userId,
+      positions,
+      window,
+      query.sync === 'observe',
+    );
     for (const position of positions) {
       const generic = metadata.get(metadataKey(position))?.data;
       const uid = position.instrumentUid || generic?.uid;
@@ -205,6 +165,85 @@ export class UpcomingEventsService {
   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): Promise<string[]> {
+    const window = dateWindow({});
+    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<ReturnType<CatalogEnrichmentService['readMany']>> =
+      [];
+    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<ReturnType<InstrumentMetadataService['savedMany']>>,
+    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 {