feat(income): add account income analytics API

src/tbank/brokerage-operations.service.ts

@@ -1,4 +1,4 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, ServiceUnavailableException } from '@nestjs/common';
 import { LRUCache } from 'lru-cache';
 import { RequestContextService } from '../common/logging/request-context.service';
 import {
@@ -12,6 +12,7 @@ import {
 import { InstrumentMetadataService } from './instrument-metadata.service';
 import { TbankAccessService } from './tbank-access.service';
 import { TbankClient } from './tbank.client';
+import { DateOnlyRange, dateOnlyBounds } from './income-period';
 
 const freshForMs = 30_000;
 const maxCachedPages = 100;
@@ -39,51 +40,123 @@ export class BrokerageOperationsService {
     this.requestContext.set({ userId, brokerageAccountId: accountId });
     const range = this.range(query);
 
-    return this.access.withAccount(userId, accountId, async ({ token }) => {
-      const key = cacheKey(userId, accountId, query, range);
-      const cached = this.pages.get(key);
-      if (cached) return cached;
+    return this.access.withAccount(userId, accountId, ({ token }) =>
+      this.page(token, userId, accountId, query, range),
+    );
+  }
 
-      if (range.from > range.to) {
-        const page = emptyPage(accountId);
-        this.pages.set(key, page);
-        return page;
-      }
-
-      const response = await this.tbank.getOperationsByCursor(token, {
-        accountId,
-        from: seconds(range.from),
-        to: seconds(range.to, /^\d{4}-\d{2}-\d{2}$/.test(query.to ?? '')),
-        cursor: query.cursor,
-        limit: query.limit ?? 100,
-        state: 'OPERATION_STATE_EXECUTED',
-      });
-      const items = (response.items ?? []).filter(
-        (item) =>
-          !query.categories?.length ||
-          query.categories.includes(operationCategory(item.type)),
-      );
-      const logoNames = new Map<string, Promise<string | null>>();
-      const page = {
-        accountId,
-        items: await Promise.all(
-          items.map(async (item) => ({
-            ...mapBrokerageOperation(item),
-            logoName: await this.logoName(
-              token,
-              item.instrumentUid,
-              item.figi,
-              logoNames,
+  async collectRanges(
+    userId: string,
+    accountId: string,
+    ranges:
+      | DateOnlyRange[]
+      | ((accountOpenedAt: Date | undefined) => DateOnlyRange[]),
+  ): Promise<{
+    accountOpenedAt: Date | undefined;
+    pages: BrokerageOperationsPageDto[][];
+  }> {
+    this.requestContext.set({ userId, brokerageAccountId: accountId });
+    return this.access.withAccount(
+      userId,
+      accountId,
+      async ({ token, account }) => {
+        const accountOpenedAt = account.openedDate?.seconds
+          ? new Date(Number(account.openedDate.seconds) * 1000)
+          : undefined;
+        const selected =
+          typeof ranges === 'function' ? ranges(accountOpenedAt) : ranges;
+        return {
+          accountOpenedAt,
+          pages: await Promise.all(
+            selected.map((range) =>
+              this.collectRange(token, userId, accountId, range),
             ),
-          })),
-        ),
-        nextCursor: response.nextCursor || null,
-        hasNext: Boolean(response.hasNext),
-        observedAt: new Date().toISOString(),
-      };
+          ),
+        };
+      },
+    );
+  }
+
+  private async collectRange(
+    token: string,
+    userId: string,
+    accountId: string,
+    range: DateOnlyRange,
+  ): Promise<BrokerageOperationsPageDto[]> {
+    const pages: BrokerageOperationsPageDto[] = [];
+    const seen = new Set<string>();
+    let cursor: string | undefined;
+    do {
+      const page = await this.page(
+        token,
+        userId,
+        accountId,
+        {
+          from: range.from,
+          cursor,
+          limit: 1000,
+          categories: ['income'],
+        },
+        dateOnlyBounds(range),
+      );
+      pages.push(page);
+      if (!page.hasNext) return pages;
+      if (!page.nextCursor || seen.has(page.nextCursor))
+        throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' });
+      seen.add(page.nextCursor);
+      cursor = page.nextCursor;
+    } while (cursor);
+    return pages;
+  }
+
+  private async page(
+    token: string,
+    userId: string,
+    accountId: string,
+    query: BrokerageOperationsQueryDto,
+    range: { from: Date; to: Date },
+  ): Promise<BrokerageOperationsPageDto> {
+    const key = cacheKey(userId, accountId, query, range);
+    const cached = this.pages.get(key);
+    if (cached) return cached;
+    if (range.from > range.to) {
+      const page = emptyPage(accountId);
       this.pages.set(key, page);
       return page;
+    }
+    const response = await this.tbank.getOperationsByCursor(token, {
+      accountId,
+      from: seconds(range.from),
+      to: seconds(range.to, /^\d{4}-\d{2}-\d{2}$/.test(query.to ?? '')),
+      cursor: query.cursor,
+      limit: query.limit ?? 100,
+      state: 'OPERATION_STATE_EXECUTED',
     });
+    const items = (response.items ?? []).filter(
+      (item) =>
+        !query.categories?.length ||
+        query.categories.includes(operationCategory(item.type)),
+    );
+    const logoNames = new Map<string, Promise<string | null>>();
+    const page = {
+      accountId,
+      items: await Promise.all(
+        items.map(async (item) => ({
+          ...mapBrokerageOperation(item),
+          logoName: await this.logoName(
+            token,
+            item.instrumentUid,
+            item.figi,
+            logoNames,
+          ),
+        })),
+      ),
+      nextCursor: response.nextCursor || null,
+      hasNext: Boolean(response.hasNext),
+      observedAt: new Date().toISOString(),
+    };
+    this.pages.set(key, page);
+    return page;
   }
 
   private range(query: BrokerageOperationsQueryDto) {