refactor: split brokerage operations repository and cursor codec (WealthLedger/backend#167) (#190)

2026-11-04 22:01:02 +03:00
parent deda7ffc2c
commit cd78b43783
9 changed files with 570 additions and 415 deletions
@@ -0,0 +1,192 @@
+import {
+  BadRequestException,
+  Injectable,
+  NotFoundException,
+} from '@nestjs/common';
+import { createHash } from 'node:crypto';
+
+import { PrismaService } from '../../../../prisma/prisma.service';
+import {
+  dateOnlyBounds,
+  DateOnlyRange,
+  moscowDate,
+} from '../../investment-calendar';
+import { OperationErrorCode } from '../domain/operation-error-code';
+import { decodeInterval } from '../domain/operations-interval-codec';
+
+const CURSOR_LOOKAHEAD_ROWS = 1;
+
+@Injectable()
+export class BrokerageOperationsReadRepository {
+  constructor(private readonly prisma: PrismaService) {}
+
+  async readPage(input: {
+    accountId: string;
+    categories: string[];
+    cursor?: { accountId: string; at: string; binding: string; id: string };
+    filterKey: string;
+    from: Date;
+    instrumentUid?: string;
+    limit?: number;
+    periods?: (openedAt: Date | undefined) => DateOnlyRange[];
+    revision: string;
+    to: Date;
+    userId: string;
+  }) {
+    return this.prisma.$transaction(async (tx) => {
+      const connection = await tx.tbankConnection.findUnique({
+        where: {
+          userId_revision: { userId: input.userId, revision: input.revision },
+          invalidToken: false,
+        },
+        select: {
+          membershipAt: true,
+          membershipRevision: true,
+          accounts: {
+            where:
+              input.accountId === 'all' ? {} : { accountId: input.accountId },
+            orderBy: { accountId: 'asc' },
+            select: {
+              accountId: true,
+              name: true,
+              openedOn: true,
+              operationsRevision: true,
+            },
+          },
+        },
+      });
+      if (!connection) return null;
+      if (
+        connection.membershipAt &&
+        input.accountId !== 'all' &&
+        !connection.accounts.length
+      ) {
+        throw new NotFoundException();
+      }
+      const accounts = connection.accounts;
+      // eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings.
+      const oldest = accounts.map((account) => account.openedOn).sort()[0];
+      const openedAt =
+        oldest && accounts.every((account) => account.openedOn)
+          ? new Date(`${oldest}T00:00:00+03:00`)
+          : undefined;
+      const periods = input.periods?.(openedAt).map((period) => {
+        const now = new Date();
+        const bounds = dateOnlyBounds(period, now);
+        return {
+          ...bounds,
+          key: `${period.from}/${period.to}`,
+          movingEnd: period.to >= moscowDate(now),
+        };
+      });
+      const binding = createHash('sha256')
+        .update(
+          JSON.stringify([
+            input.userId,
+            input.revision,
+            input.accountId === 'all' ? connection.membershipRevision : null,
+            accounts.map((account) => [
+              account.accountId,
+              account.operationsRevision,
+            ]),
+            input.filterKey,
+          ]),
+        )
+        .digest('base64url');
+      if (input.cursor && input.cursor.binding !== binding) {
+        throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor });
+      }
+      const scope = {
+        userId: input.userId,
+        revision: input.revision,
+        accountId: { in: accounts.map((account) => account.accountId) },
+      };
+      const coverage = await tx.tbankOperationCoverage.findMany({
+        where: scope,
+      });
+      const work = await tx.tbankSyncWork.findMany({
+        where: {
+          userId: input.userId,
+          revision: input.revision,
+          scopeKey: { startsWith: 'operations:' },
+          state: { in: ['running', 'failed'] },
+        },
+        select: {
+          id: true,
+          state: true,
+          errorCode: true,
+          intent: true,
+          priority: true,
+        },
+      });
+      const membershipWork = await tx.tbankSyncWork.findUnique({
+        where: {
+          userId_revision_scopeKey: {
+            userId: input.userId,
+            revision: input.revision,
+            scopeKey: 'membership',
+          },
+        },
+        select: { id: true, state: true, errorCode: true },
+      });
+      const after = input.cursor;
+      const rows = await tx.tbankOperation.findMany({
+        where: {
+          ...scope,
+          ...(periods
+            ? {
+                OR: periods.map((period) => ({
+                  occurredAt: { gte: period.from, lt: period.to },
+                })),
+              }
+            : { occurredAt: { gte: input.from, lt: input.to } }),
+          ...(input.instrumentUid
+            ? { instrumentUid: input.instrumentUid }
+            : {}),
+          ...(input.categories.length
+            ? { category: { in: input.categories } }
+            : {}),
+          ...(after
+            ? {
+                OR: [
+                  { occurredAt: { lt: new Date(after.at) } },
+                  {
+                    occurredAt: new Date(after.at),
+                    accountId: { gt: after.accountId },
+                  },
+                  {
+                    occurredAt: new Date(after.at),
+                    accountId: after.accountId,
+                    brokerOperationId: { gt: after.id },
+                  },
+                ],
+              }
+            : {}),
+        },
+        orderBy: [
+          { occurredAt: 'desc' },
+          { accountId: 'asc' },
+          { brokerOperationId: 'asc' },
+        ],
+        ...(input.limit === undefined ? {} : { take: input.limit + CURSOR_LOOKAHEAD_ROWS }),
+      });
+      return {
+        connection,
+        periods,
+        membershipWork,
+        binding,
+        rows,
+        coverage: coverage.map(({ instrumentPredicate, ...row }) => ({
+          ...row,
+          instrumentUid: instrumentPredicate || undefined,
+        })),
+        work: work.map(({ intent, ...row }) => ({
+          ...row,
+          intervals: (intent as { intervals: string[] }).intervals.map(
+            decodeInterval,
+          ),
+        })),
+      };
+    });
+  }
+}
@@ -1,8 +1,4 @@
-import {
-  BadRequestException,
-  Injectable,
-  NotFoundException,
-} from '@nestjs/common';
+import { Injectable, NotFoundException } from '@nestjs/common';
 import { createHash, randomUUID } from 'node:crypto';
 import { isDeepStrictEqual } from 'node:util';

@@ -17,11 +13,11 @@ import {
 } from '../../../../integrations/tbank/tbank.api';
 import { PrismaService } from '../../../../prisma/prisma.service';
 import {
-  dateOnlyBounds,
-  DateOnlyRange,
-  moscowDate,
-} from '../../investment-calendar';
-import { OperationErrorCode } from '../domain/operation-error-code';
+  decodeInterval,
+  extendRunningIntent,
+  intervalKey,
+  type OperationsIntent,
+} from '../domain/operations-interval-codec';
 import {
  type OperationInterval,
  operationIntervals,
@@ -34,13 +30,13 @@ import {
  operationTimestamp,
 } from './brokerage-operation.mapper';

-const CURSOR_LOOKAHEAD_ROWS = 1;
 const EMPTY_INTERVAL_COUNT = 0;
 const SINGLE_WORK_GROUP = 1;
 const INITIAL_PAGE_COUNT = 0;
 const PAGE_INCREMENT = 1;
 const EXPECTED_UPDATED_WORK_ITEMS = 1;

+export type { OperationsIntent } from '../domain/operations-interval-codec';
 export type { OperationInterval } from '../domain/operations-sync-plan';

 export type OperationAttempt = {
@@ -49,12 +45,6 @@ export type OperationAttempt = {
  revision: string;
  userId: string;
 };
-type OperationsIntent = {
-  atomic: boolean;
-  atomicIntervals?: string[];
-  intervals: string[];
-  screenIntervals?: string[];
-};
 type PageProgress = {
  complete: boolean;
  cursor: string;
@@ -67,176 +57,6 @@ const progressItem = '';
 export class BrokerageOperationsRepository {
  constructor(private readonly prisma: PrismaService) {}

-  async readPage(input: {
-    accountId: string;
-    categories: string[];
-    cursor?: { accountId: string; at: string; binding: string; id: string };
-    filterKey: string;
-    from: Date;
-    instrumentUid?: string;
-    limit?: number;
-    periods?: (openedAt: Date | undefined) => DateOnlyRange[];
-    revision: string;
-    to: Date;
-    userId: string;
-  }) {
-    return this.prisma.$transaction(async (tx) => {
-      const connection = await tx.tbankConnection.findUnique({
-        where: {
-          userId_revision: { userId: input.userId, revision: input.revision },
-          invalidToken: false,
-        },
-        select: {
-          membershipAt: true,
-          membershipRevision: true,
-          accounts: {
-            where:
-              input.accountId === 'all' ? {} : { accountId: input.accountId },
-            orderBy: { accountId: 'asc' },
-            select: {
-              accountId: true,
-              name: true,
-              openedOn: true,
-              operationsRevision: true,
-            },
-          },
-        },
-      });
-      if (!connection) return null;
-      if (
-        connection.membershipAt &&
-        input.accountId !== 'all' &&
-        !connection.accounts.length
-      ) {
-        throw new NotFoundException();
-      }
-      const accounts = connection.accounts;
-      // eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings.
-      const oldest = accounts.map((account) => account.openedOn).sort()[0];
-      const openedAt =
-        oldest && accounts.every((account) => account.openedOn)
-          ? new Date(`${oldest}T00:00:00+03:00`)
-          : undefined;
-      const periods = input.periods?.(openedAt).map((period) => {
-        const now = new Date();
-        const bounds = dateOnlyBounds(period, now);
-        return {
-          ...bounds,
-          key: `${period.from}/${period.to}`,
-          movingEnd: period.to >= moscowDate(now),
-        };
-      });
-      const binding = createHash('sha256')
-        .update(
-          JSON.stringify([
-            input.userId,
-            input.revision,
-            input.accountId === 'all' ? connection.membershipRevision : null,
-            accounts.map((account) => [
-              account.accountId,
-              account.operationsRevision,
-            ]),
-            input.filterKey,
-          ]),
-        )
-        .digest('base64url');
-      if (input.cursor && input.cursor.binding !== binding) {
-        throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor });
-      }
-      const scope = {
-        userId: input.userId,
-        revision: input.revision,
-        accountId: { in: accounts.map((account) => account.accountId) },
-      };
-      const coverage = await tx.tbankOperationCoverage.findMany({
-        where: scope,
-      });
-      const work = await tx.tbankSyncWork.findMany({
-        where: {
-          userId: input.userId,
-          revision: input.revision,
-          scopeKey: { startsWith: 'operations:' },
-          state: { in: ['running', 'failed'] },
-        },
-        select: {
-          id: true,
-          state: true,
-          errorCode: true,
-          intent: true,
-          priority: true,
-        },
-      });
-      const membershipWork = await tx.tbankSyncWork.findUnique({
-        where: {
-          userId_revision_scopeKey: {
-            userId: input.userId,
-            revision: input.revision,
-            scopeKey: 'membership',
-          },
-        },
-        select: { id: true, state: true, errorCode: true },
-      });
-      const after = input.cursor;
-      const rows = await tx.tbankOperation.findMany({
-        where: {
-          ...scope,
-          ...(periods
-            ? {
-                OR: periods.map((period) => ({
-                  occurredAt: { gte: period.from, lt: period.to },
-                })),
-              }
-            : { occurredAt: { gte: input.from, lt: input.to } }),
-          ...(input.instrumentUid
-            ? { instrumentUid: input.instrumentUid }
-            : {}),
-          ...(input.categories.length
-            ? { category: { in: input.categories } }
-            : {}),
-          ...(after
-            ? {
-                OR: [
-                  { occurredAt: { lt: new Date(after.at) } },
-                  {
-                    occurredAt: new Date(after.at),
-                    accountId: { gt: after.accountId },
-                  },
-                  {
-                    occurredAt: new Date(after.at),
-                    accountId: after.accountId,
-                    brokerOperationId: { gt: after.id },
-                  },
-                ],
-              }
-            : {}),
-        },
-        orderBy: [
-          { occurredAt: 'desc' },
-          { accountId: 'asc' },
-          { brokerOperationId: 'asc' },
-        ],
-        ...(input.limit === undefined ? {} : { take: input.limit + CURSOR_LOOKAHEAD_ROWS }),
-      });
-      return {
-        connection,
-        periods,
-        membershipWork,
-        binding,
-        rows,
-        coverage: coverage.map(({ instrumentPredicate, ...row }) => ({
-          ...row,
-          instrumentUid: instrumentPredicate || undefined,
-        })),
-        work: work.map(({ intent, ...row }) => ({
-          ...row,
-          intervals: (intent as { intervals: string[] }).intervals.map(
-            decodeInterval,
-          ),
-        })),
-      };
-    });
-  }
-
  async coverage(userId: string, revision: string, accountIds: string[]) {
    const rows = await this.prisma.tbankOperationCoverage.findMany({
      where: {
@@ -875,53 +695,6 @@ export class BrokerageOperationsRepository {
  }
 }

-function extendRunningIntent(input: {
-  additions: OperationInterval[];
-  intervals: OperationInterval[];
-  pending: OperationInterval[];
-  previous: OperationsIntent | undefined;
-  previousPriority: number;
-  priority: number;
-  publicationAtomic: boolean;
-}): OperationsIntent {
-  const { previous, pending, additions, intervals, priority, previousPriority, publicationAtomic } = input;
-  const overlapsRequested = (item: OperationInterval) =>
-    intervals.some((target) => operationIntervalsOverlap(item, target));
-  const retainedScreen = previous?.screenIntervals ??
-    (previousPriority === operationsSyncPriority.screen ? (previous?.intervals ?? []) : []);
-  const addedScreen = priority === operationsSyncPriority.screen
-    ? [...pending, ...additions].filter(overlapsRequested).map(intervalKey)
-    : [];
-  const ordered = priority < previousPriority
-    ? [
-        ...pending.filter(overlapsRequested),
-        ...additions,
-        ...pending.filter((item) => !overlapsRequested(item)),
-      ]
-    : [...pending, ...additions];
-  return {
-    atomic: previous?.atomic ?? publicationAtomic,
-    atomicIntervals: previous?.atomic
-      ? [...new Set([...(previous.atomicIntervals ?? previous.intervals), ...additions.map(intervalKey)])]
-      : [],
-    screenIntervals: [...new Set([...retainedScreen, ...addedScreen])],
-    intervals: ordered.map(intervalKey),
-  };
-}
-
-function intervalKey(interval: OperationInterval): string {
-  // eslint-disable-next-line sonarjs/no-inverted-boolean-check -- Negation also rejects Invalid Date; >= would silently accept NaN.
-  if (!(interval.from < interval.to)) {
-    throw new Error('Invalid operations interval');
-  }
-  return JSON.stringify([
-    interval.accountId,
-    interval.from.toISOString(),
-    interval.to.toISOString(),
-    interval.instrumentUid ?? '',
-  ]);
-}
-
 async function claim(
  tx: Prisma.TransactionClient,
  attempt: OperationAttempt,
@@ -929,13 +702,3 @@ async function claim(
  const count = await claimSyncWork(tx, attempt, { updatedAt: new Date() });
  return count === EXPECTED_UPDATED_WORK_ITEMS;
 }
-
-function decodeInterval(key: string): OperationInterval {
-  const [accountId, from, to, instrumentUid] = JSON.parse(key) as string[];
-  return {
-    accountId,
-    from: new Date(from),
-    to: new Date(to),
-    ...(instrumentUid ? { instrumentUid } : {}),
-  };
-}