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

src/modules/investments/operations/application/brokerage-operations.service.ts

@@ -1,6 +1,5 @@
-import { BadRequestException, Injectable } from '@nestjs/common';
-import { ConfigService } from '@nestjs/config';
-import { createHash, createHmac, timingSafeEqual } from 'node:crypto';
+import { Injectable } from '@nestjs/common';
+import { createHash } from 'node:crypto';
 
 import {
   InstrumentMetadataService,
@@ -11,51 +10,47 @@ import {
 } from '../../../../integrations/tbank/tbank.api';
 import { dateOnlyBounds, DateOnlyRange } from '../../investment-calendar';
 import { investmentResourceMeta } from '../../investment-resource.mapper';
-import { accountHistory, ledgerRange } from '../../investment-sync-plan';
+import { ledgerRange } from '../../investment-sync-plan';
 import {
   hourlyAttempt,
   TbankSyncRepository,
 } from '../../tbank-sync/tbank-sync.api';
-import { OperationErrorCode } from '../domain/operation-error-code';
+import {
+  coverageAsOf,
+  deriveReadCoverage,
+  historyState,
+  serializeRange,
+} from '../domain/operations-coverage';
 import {
   confirmedOperationRanges,
   operationIntervalsOverlap,
-  operationsSyncPriority,
   planOperationRead,
   uncoveredOperationIntervals,
 } from '../domain/operations-sync-plan';
 import { mapBrokerageOperation } from '../infrastructure/brokerage-operation.mapper';
-import { BrokerageOperationsRepository } from '../infrastructure/brokerage-operations.repository';
+import { BrokerageOperationsReadRepository } from '../infrastructure/brokerage-operations-read.repository';
 import {
   BrokerageOperationsQueryDto,
   DEFAULT_OPERATIONS_PAGE_SIZE,
 } from '../presentation/brokerage-operation.dto';
 import { BrokerageOperationsSyncService } from './brokerage-operations-sync.service';
+import { LocalOperationsCursorCodec } from './local-operations-cursor.codec';
 
 const FIRST_RESULT_OFFSET = 0;
 const MISSING_MEMBERSHIP_REVISION = 0;
 const STALE_FRESHNESS_MS = 0;
-const UNKNOWN_AS_OF_MS = 0;
+const NO_COVERAGE_ROWS = 0;
 const LAST_ROW_INDEX = -1;
 const UNIX_EPOCH_MS = 0;
 
-type LocalOperationsCursor = {
-  accountId: string;
-  at: string;
-  binding: string;
-  id: string;
-  movingEnd?: boolean;
-  to: string;
-};
-
 @Injectable()
 export class BrokerageOperationsService {
   constructor(
     private readonly access: TbankAccessService,
     private readonly metadata: InstrumentMetadataService,
-    private readonly repository: BrokerageOperationsRepository,
+    private readonly repository: BrokerageOperationsReadRepository,
     private readonly sync: BrokerageOperationsSyncService,
-    private readonly config: ConfigService,
+    private readonly cursorCodec: LocalOperationsCursorCodec,
     private readonly hourlySync: TbankSyncRepository,
   ) {}
 
@@ -65,7 +60,7 @@ export class BrokerageOperationsService {
     query: BrokerageOperationsQueryDto,
   ) {
     const connection = await this.access.local(userId, accountId);
-    const cursor = query.cursor ? this.decodeCursor(query.cursor) : undefined;
+    const cursor = query.cursor ? this.cursorCodec.decode(query.cursor) : undefined;
     const selected = connection.accounts.filter(
       (account) => accountId === 'all' || account.accountId === accountId,
     );
@@ -124,64 +119,18 @@ export class BrokerageOperationsService {
     const saved = await this.repository.readPage(readInput);
     const hourly = await this.hourlySync.hourly(userId);
     const accounts = saved?.connection.accounts ?? [];
-    const targets = accounts.flatMap((account) =>
-      from < to
-        ? [
-            {
-              accountId: account.accountId,
-              from,
-              to,
-              instrumentUid: query.instrumentUid,
-            },
-          ]
-        : [],
-    );
-    const plan = planOperationRead(targets, saved?.coverage ?? [], now, {
+    const view = deriveReadCoverage({
+      accounts,
+      connectionMembershipAt: saved?.connection.membershipAt ?? null,
+      coverage: saved?.coverage ?? [],
+      from,
+      instrumentUid: query.instrumentUid,
       movingEnd,
-    });
-    const confirmed = confirmedOperationRanges(plan.intervals, saved?.coverage ?? []);
-    const missing = uncoveredOperationIntervals(
-      plan.intervals,
-      confirmed.map((range) => ({
-        ...range,
-        instrumentUid: query.instrumentUid,
-      })),
-    );
-    const available = Boolean(saved?.connection.membershipAt);
-    const complete = available && !missing.length;
-    const relevant =
-      saved?.work.filter((work) =>
-        work.intervals.some((interval) =>
-          targets.some((target) => operationIntervalsOverlap(interval, target)),
-        ),
-      ) ?? [];
-    const work =
-      relevant.find((work) => work.state === 'running') ??
-      relevant.find((work) => work.state === 'failed') ??
-      null;
-    const sources = (saved?.coverage ?? []).filter((row) =>
-      targets.some(
-        (target) =>
-          operationIntervalsOverlap(row, target) &&
-          (!row.instrumentUid || row.instrumentUid === target.instrumentUid),
-      ),
-    );
-    const lastSuccessfulAt =
-      sources
-        .map((row) => row.fetchedAt)
-        .sort((a, b) => a.valueOf() - b.valueOf())[0] ??
-        (complete ? saved?.connection.membershipAt : null) ??
-        null;
-    const historyTargets = accountHistory(accounts, now);
-    const history = planOperationRead(
-      historyTargets,
-      saved?.coverage ?? [],
       now,
-      { movingEnd: true },
-    );
-    const knownBoundary = accounts.every(
-      (account) => account.openedOn !== null,
-    );
+      to,
+      work: saved?.work ?? [],
+    });
+    const { targets, plan, confirmed, missing, available, complete, lastSuccessfulAt, history, knownBoundary, work } = view;
     const rows = saved?.rows.slice(FIRST_RESULT_OFFSET, limit) ?? [];
     const instruments = await this.metadata.savedMany(
       rows.map((row) => row.payload as TbankOperationItem),
@@ -224,49 +173,17 @@ export class BrokerageOperationsService {
       lastAttempt: hourlyAttempt(hourly, 'operations', accountId === 'all' ? undefined : accountId),
       dependencies: [membershipMeta, ...metadataDependencies],
       coverage: {
-        requested: targets.map(serialize),
-        completeRanges: confirmed.map(serialize),
-        missingRanges: missing.map(serialize),
-        asOf:
-          accounts.length &&
-          accounts.every((account) =>
-            plan.intervals.some(
-              (interval) =>
-                interval.accountId === account.accountId && interval.asOf,
-            ),
-          )
-            ? new Date(
-                Math.min(
-                  ...accounts.map((account) =>
-                    Math.max(
-                      ...plan.intervals
-                        .filter(
-                          (interval) =>
-                            interval.accountId === account.accountId,
-                        )
-                        .map((interval) => interval.asOf?.valueOf() ?? UNKNOWN_AS_OF_MS),
-                    ),
-                  ),
-                ),
-              ).toISOString()
-            : null,
+        requested: targets.map(serializeRange),
+        completeRanges: confirmed.map(serializeRange),
+        missingRanges: missing.map(serializeRange),
+        asOf: coverageAsOf(plan, accounts.map((account) => account.accountId)),
         historyBoundary: knownBoundary ? 'known' : 'unknown',
-        historyState: saved?.work.some(
-          (work) => work.state === 'running' && work.priority > operationsSyncPriority.screen,
-        )
-          ? 'loading'
-          : saved?.work.some(
-            (work) => work.state === 'failed' && work.priority > operationsSyncPriority.screen,
-          )
-            ? 'failed'
-            : knownBoundary &&
-              history.intervals.every(
-                (interval) => interval.availability === 'complete',
-              )
-              ? 'complete'
-              : saved?.coverage.length
-                ? 'partial'
-                : 'missing',
+        historyState: historyState(
+          knownBoundary,
+          history,
+          saved?.work ?? [],
+          saved?.coverage.length ?? NO_COVERAGE_ROWS,
+        ),
       },
     });
     const last = rows.at(LAST_ROW_INDEX);
@@ -301,7 +218,7 @@ export class BrokerageOperationsService {
               hasNext,
               nextCursor:
                 hasNext && last && saved
-                  ? this.encodeCursor({
+                  ? this.cursorCodec.encode({
                       binding: saved.binding,
                       movingEnd,
                       to: to.toISOString(),
@@ -316,50 +233,6 @@ export class BrokerageOperationsService {
     };
   }
 
-  private encodeCursor(value: LocalOperationsCursor): string {
-    const body = Buffer.from(JSON.stringify(value)).toString('base64url');
-    return `${body}.${createHmac('sha256', this.config.getOrThrow<string>('JWT_SECRET')).update(body).digest('base64url')}`;
-  }
-
-  private decodeCursor(value: string): LocalOperationsCursor {
-    try {
-      const [body, signature, extra] = value.split('.');
-      const expected = createHmac(
-        'sha256',
-        this.config.getOrThrow<string>('JWT_SECRET'),
-      )
-        .update(body)
-        .digest();
-      const actual = Buffer.from(signature ?? '', 'base64url');
-      if (
-        extra ||
-        actual.length !== expected.length ||
-        !timingSafeEqual(actual, expected)
-      ) {
-        throw new Error();
-      }
-      const cursor = JSON.parse(
-        Buffer.from(body, 'base64url').toString(),
-      ) as Partial<LocalOperationsCursor>;
-      if (
-        (cursor.movingEnd !== undefined &&
-          typeof cursor.movingEnd !== 'boolean') ||
-          !cursor.binding ||
-          !cursor.accountId ||
-          !cursor.id ||
-          !cursor.at ||
-          !cursor.to ||
-          !Number.isFinite(Date.parse(cursor.at)) ||
-          !Number.isFinite(Date.parse(cursor.to))
-      ) {
-        throw new Error();
-      }
-      return cursor as LocalOperationsCursor;
-    } catch {
-      throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor });
-    }
-  }
-
   async collectRanges(
     userId: string,
     accountId: string,
@@ -466,12 +339,12 @@ export class BrokerageOperationsService {
         lastAttempt: hourlyAttempt(hourly, 'operations', accountId === 'all' ? undefined : accountId),
         dependencies: [membership],
         coverage: {
-          requested: targets.map(serialize),
-          completeRanges: confirmed.map(serialize),
+          requested: targets.map(serializeRange),
+          completeRanges: confirmed.map(serializeRange),
           missingRanges: uncoveredOperationIntervals(
             plan.intervals,
             confirmed,
-          ).map(serialize),
+          ).map(serializeRange),
           asOf:
             complete && targets.length
               ? new Date(
@@ -541,7 +414,3 @@ export class BrokerageOperationsService {
     });
   }
 }
-
-function serialize(range: { accountId: string; from: Date; to: Date }) {
-  return { accountId: range.accountId, from: range.from.toISOString(), to: range.to.toISOString() };
-}

src/modules/investments/operations/application/local-operations-cursor.codec.spec.ts

@@ -0,0 +1,23 @@
+import { decode, encode } from './local-operations-cursor.codec';
+
+const secret = 'secret';
+const cursor = {
+  accountId: 'account',
+  at: '2026-01-01T00:00:00.000Z',
+  binding: 'owner-filter-binding',
+  id: 'operation',
+  movingEnd: false,
+  to: '2026-02-01T00:00:00.000Z',
+};
+
+describe('local operations cursor', () => {
+  it('roundtrips the opaque cursor, including its owner and filter binding', () => {
+    expect(decode(encode(cursor, secret), secret)).toEqual(cursor);
+  });
+
+  it('rejects a tampered cursor and a cursor signed with a different key', () => {
+    const encoded = encode(cursor, secret);
+    expect(() => decode(`${encoded}x`, secret)).toThrow();
+    expect(() => decode(encoded, 'different-secret')).toThrow();
+  });
+});

src/modules/investments/operations/application/local-operations-cursor.codec.ts

@@ -0,0 +1,72 @@
+import { BadRequestException, Injectable } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { createHmac, timingSafeEqual } from 'node:crypto';
+
+import { OperationErrorCode } from '../domain/operation-error-code';
+
+export type LocalOperationsCursor = {
+  accountId: string;
+  at: string;
+  binding: string;
+  id: string;
+  movingEnd?: boolean;
+  to: string;
+};
+
+// Opaque cursor: base64url body + HMAC-SHA256 signature over JWT_SECRET.
+// Wire format and validation are part of the HTTP contract and must not change.
+export function encode(value: LocalOperationsCursor, secret: string): string {
+  const body = Buffer.from(JSON.stringify(value)).toString('base64url');
+  return `${body}.${createHmac('sha256', secret).update(body).digest('base64url')}`;
+}
+
+export function decode(value: string, secret: string): LocalOperationsCursor {
+  try {
+    const [body, signature, extra] = value.split('.');
+    const expected = createHmac('sha256', secret).update(body).digest();
+    const actual = Buffer.from(signature ?? '', 'base64url');
+    if (
+      extra ||
+      actual.length !== expected.length ||
+      !timingSafeEqual(actual, expected)
+    ) {
+      throw new Error();
+    }
+    const cursor = JSON.parse(
+      Buffer.from(body, 'base64url').toString(),
+    ) as Partial<LocalOperationsCursor>;
+    if (
+      (cursor.movingEnd !== undefined &&
+        typeof cursor.movingEnd !== 'boolean') ||
+        !cursor.binding ||
+        !cursor.accountId ||
+        !cursor.id ||
+        !cursor.at ||
+        !cursor.to ||
+        !Number.isFinite(Date.parse(cursor.at)) ||
+        !Number.isFinite(Date.parse(cursor.to))
+    ) {
+      throw new Error();
+    }
+    return cursor as LocalOperationsCursor;
+  } catch {
+    throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor });
+  }
+}
+
+@Injectable()
+export class LocalOperationsCursorCodec {
+  constructor(private readonly config: ConfigService) {}
+
+  encode(value: LocalOperationsCursor): string {
+    return encode(value, this.secret());
+  }
+
+  decode(value: string): LocalOperationsCursor {
+    return decode(value, this.secret());
+  }
+
+  private secret(): string {
+    return this.config.getOrThrow<string>('JWT_SECRET');
+  }
+}