refactor: share prisma json codec and check progress shape (WealthLedger/backend#170)

src/modules/investments/instruments/infrastructure/candle.repository.ts

@@ -5,6 +5,7 @@ import type { Prisma } from '../../../../../generated/prisma/client';
 import type { TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
 
 import { PrismaService } from '../../../../prisma/prisma.service';
+import { toJson } from '../../../../prisma/to-json';
 import {
   type CandleRange,
   confirmedCandleCoverage,
@@ -115,7 +116,7 @@ export class CandleRepository {
         processRunId: runId,
         state: 'running',
         priority,
-        intent: JSON.parse(JSON.stringify(intent)) as Prisma.InputJsonObject,
+        intent: toJson(intent),
         progress: {},
         errorCode: null,
       };
@@ -178,9 +179,7 @@ export class CandleRepository {
         sourceKey: 'candles',
         itemKey: key,
       };
-      const payload = JSON.parse(
-        JSON.stringify({ range, candles: rows, fetchedAt }),
-      ) as Prisma.InputJsonObject;
+      const payload = toJson({ range, candles: rows, fetchedAt });
       await tx.tbankSyncStage.upsert({
         where: { workId_attemptId_sourceKey_itemKey: identity },
         create: { ...identity, payload },

src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts

@@ -2,11 +2,10 @@ import { Injectable } from '@nestjs/common';
 import { randomUUID } from 'node:crypto';
 import { isDeepStrictEqual } from 'node:util';
 
-import type { Prisma } from '../../../../../generated/prisma/client';
-
 import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
 import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
 import { PrismaService } from '../../../../prisma/prisma.service';
+import { toJson } from '../../../../prisma/to-json';
 
 const UNCHANGED_REVISION_INCREMENT = 0;
 const REVISION_INCREMENT = 1;
@@ -177,9 +176,7 @@ export class CatalogDiscoveryRepository {
         data: { state: 'complete', progress: {} },
       });
       if (!claimed.count) return false;
-      const payload = JSON.parse(
-        JSON.stringify(value),
-      ) as Prisma.InputJsonObject;
+      const payload = toJson(value);
       const identity = { provider, uid: input.uid };
       await tx.tbankInstrument.upsert({
         where: { provider_uid: identity },
@@ -202,12 +199,10 @@ export class CatalogDiscoveryRepository {
         const previous = await tx.tbankInstrument.findUniqueOrThrow({
           where: { provider_uid: identity },
         });
-        const metadata = JSON.parse(
-          JSON.stringify({
-            ...instrument,
-            instrumentType: input.instrumentType,
-          }),
-        ) as Prisma.InputJsonObject;
+        const metadata = toJson({
+          ...instrument,
+          instrumentType: input.instrumentType,
+        });
         await tx.tbankInstrument.update({
           where: { provider_uid: identity },
           data: {
@@ -353,22 +348,20 @@ export class CatalogDiscoveryRepository {
       for (const instrument of instruments) {
         // Discovery can create a minimal identity, but cannot overwrite verified metadata.
         const identity = { provider, uid: instrument.uid };
-        const payload = JSON.parse(
-          JSON.stringify({
-            uid: instrument.uid,
-            figi: instrument.figi,
-            ticker: instrument.ticker,
-            name: instrument.name,
-            isin: instrument.isin,
-            classCode: instrument.classCode,
-            instrumentKind: instrument.instrumentKind,
-            apiTradeAvailableFlag: instrument.apiTradeAvailableFlag,
-            currency: instrument.currency,
-            brand: instrument.brand
-              ? { logoName: instrument.brand.logoName }
-              : undefined,
-          }),
-        ) as Prisma.InputJsonObject;
+        const payload = toJson({
+          uid: instrument.uid,
+          figi: instrument.figi,
+          ticker: instrument.ticker,
+          name: instrument.name,
+          isin: instrument.isin,
+          classCode: instrument.classCode,
+          instrumentKind: instrument.instrumentKind,
+          apiTradeAvailableFlag: instrument.apiTradeAvailableFlag,
+          currency: instrument.currency,
+          brand: instrument.brand
+            ? { logoName: instrument.brand.logoName }
+            : undefined,
+        });
         await tx.tbankInstrument.upsert({
           where: { provider_uid: identity },
           update: {},

src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts

@@ -15,6 +15,7 @@ import type {
 import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
 import { isTbankProtocolValue, TbankOperationState } from '../../../../integrations/tbank/tbank-protocol';
 import { PrismaService } from '../../../../prisma/prisma.service';
+import { toJson } from '../../../../prisma/to-json';
 import {
   dateOnlyBounds,
   DateOnlyRange,
@@ -656,9 +657,7 @@ export class BrokerageOperationsRepository {
         ) {
           continue;
         }
-        const payload = JSON.parse(
-          JSON.stringify({ ...operation, cursor: undefined }),
-        ) as Prisma.InputJsonObject;
+        const payload = toJson({ ...operation, cursor: undefined });
         // Provider pagination position is not part of a financial row.
         const identity = { ...stage, itemKey: operation.id };
         const previous = await tx.tbankSyncStage.findUnique({

src/modules/investments/portfolio/infrastructure/forecast.repository.ts

@@ -4,6 +4,7 @@ import { isDeepStrictEqual } from 'node:util';
 import type { Prisma } from '../../../../../generated/prisma/client';
 
 import { PrismaService } from '../../../../prisma/prisma.service';
+import { toJson } from '../../../../prisma/to-json';
 
 const FIRST_BATCH_OFFSET = 0;
 
@@ -142,13 +143,11 @@ export class ForecastRepository {
       ) {
         return fallback;
       }
-      const revisions = JSON.parse(
-        JSON.stringify({
-          portfolioProjection: input.portfolioProjection,
-          portfolios: input.portfolios,
-          sources: input.sources,
-        }),
-      ) as Prisma.InputJsonObject;
+      const revisions = toJson({
+        portfolioProjection: input.portfolioProjection,
+        portfolios: input.portfolios,
+        sources: input.sources,
+      });
       const timestamps = input.timestamps as Prisma.InputJsonArray;
       if (
         fallback?.scopeKey === input.scopeKey &&
@@ -159,9 +158,7 @@ export class ForecastRepository {
       }
       const data = {
         membershipRevision: input.membershipRevision,
-        payload: JSON.parse(
-          JSON.stringify(input.candidate),
-        ) as Prisma.InputJsonObject,
+        payload: toJson(input.candidate),
         dependencyRevisions: revisions,
         sourceTimestamps: timestamps,
         calculatedAt: new Date(),

src/modules/investments/portfolio/infrastructure/portfolio.repository.ts

@@ -7,6 +7,7 @@ import type { PortfolioDto } from '../presentation/portfolio.dto';
 import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
 import { tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api';
 import { PrismaService } from '../../../../prisma/prisma.service';
+import { toJson } from '../../../../prisma/to-json';
 
 const EXPECTED_UPDATED_ACCOUNTS = 1;
 
@@ -137,7 +138,7 @@ export class PortfolioRepository {
         accountId,
       };
       const data = {
-        payload: JSON.parse(JSON.stringify(snapshot)) as Prisma.InputJsonObject,
+        payload: toJson(snapshot),
         fetchedAt: new Date(snapshot.lastSuccessfulAt),
       };
       await tx.tbankPortfolio.upsert({

src/modules/investments/tbank-sync/tbank-sync.repository.spec.ts

@@ -2,7 +2,11 @@ jest.mock('../../../prisma/prisma.service', () => ({
   PrismaService: class {},
 }));
 
-import { screenIntent } from './tbank-sync.repository';
+import {
+  hourlyProgress,
+  screenIntent,
+  screenProgress,
+} from './tbank-sync.repository';
 
 describe('screen sync intent', () => {
   it('restores a persisted intent and rejects a corrupted one', () => {
@@ -12,3 +16,66 @@ describe('screen sync intent', () => {
     expect(() => screenIntent(null)).toThrow('Corrupted screen sync intent');
   });
 });
+
+describe('screen sync progress', () => {
+  it('restores a persisted progress unchanged', () => {
+    const progress = { attempts: { 'work-1': 'attempt-1' }, children: ['work-1'], dispatched: true };
+    expect(screenProgress(progress)).toEqual(progress);
+  });
+
+  it('tolerates legacy rows without attempts', () => {
+    expect(screenProgress({ children: [], dispatched: false })).toEqual({
+      children: [],
+      dispatched: false,
+    });
+  });
+
+  it('rejects corrupt rows with an explicit error', () => {
+    expect(() => screenProgress(null)).toThrow('Corrupted screen sync progress');
+    expect(() => screenProgress({})).toThrow('Corrupted screen sync progress');
+    expect(() => screenProgress({ children: 'x', dispatched: false })).toThrow(
+      'Corrupted screen sync progress',
+    );
+    expect(() => screenProgress({ children: [1], dispatched: false })).toThrow(
+      'Corrupted screen sync progress',
+    );
+    expect(
+      () => screenProgress({ children: [], dispatched: 'yes' }),
+    ).toThrow('Corrupted screen sync progress');
+    expect(
+      () => screenProgress({ children: [], dispatched: false, attempts: { w: 7 } }),
+    ).toThrow('Corrupted screen sync progress');
+  });
+});
+
+describe('hourly sync progress', () => {
+  it('restores a persisted progress unchanged', () => {
+    const progress = {
+      attempts: { 'work-1': 'attempt-1' },
+      children: [{ id: 'work-1', section: 'portfolio' }],
+      result: 'partial',
+      sections: [{ errorCode: null, section: 'events' }],
+    };
+    expect(hourlyProgress(progress)).toEqual(progress);
+  });
+
+  it('tolerates legacy rows without optional collections', () => {
+    expect(hourlyProgress({ sections: [] })).toEqual({ sections: [] });
+    // Legacy hourly rows were seeded with an empty progress object.
+    expect(hourlyProgress({})).toEqual({});
+  });
+
+  it('rejects corrupt rows with an explicit error', () => {
+    expect(() => hourlyProgress(null)).toThrow('Corrupted hourly sync progress');
+    expect(() => hourlyProgress({ sections: 'x' })).toThrow('Corrupted hourly sync progress');
+    expect(() => hourlyProgress({ sections: [{ errorCode: 'x', section: 1 }] })).toThrow(
+      'Corrupted hourly sync progress',
+    );
+    expect(() =>
+      hourlyProgress({ sections: [], children: [{ id: 'w', section: 5 }] }),
+    ).toThrow('Corrupted hourly sync progress');
+    expect(() => hourlyProgress({ sections: [], attempts: { w: 7 } })).toThrow(
+      'Corrupted hourly sync progress',
+    );
+  });
+});

src/modules/investments/tbank-sync/tbank-sync.repository.ts

@@ -7,6 +7,7 @@ import type { InvestmentAttemptDto } from '../investment-resource.dto';
 import type { InvestmentScreenDto } from './tbank-sync.dto';
 
 import { PrismaService } from '../../../prisma/prisma.service';
+import { toJson } from '../../../prisma/to-json';
 export type ScreenIntent = { now: string; scope: InvestmentScreenDto };
 
 // Json columns are written only by this repository; a shape check turns corruption into an explicit 500.
@@ -22,6 +23,31 @@ export type ScreenProgress = {
   children: string[];
   dispatched: boolean;
 };
+// Progress columns are written only by this repository; a shape check turns
+// corruption into an explicit 500. Legacy rows predate optional fields, so
+// only the fields every writer has always written are required.
+function validAttemptMap(attempts: unknown): boolean {
+  return (
+    attempts === undefined ||
+    (typeof attempts === 'object' &&
+      attempts !== null &&
+      !Array.isArray(attempts) &&
+      Object.values(attempts).every((id) => typeof id === 'string'))
+  );
+}
+
+export function screenProgress(value: Prisma.JsonValue): ScreenProgress {
+  const progress = value as null | Partial<ScreenProgress>;
+  if (
+    typeof progress?.dispatched !== 'boolean' ||
+    !Array.isArray(progress.children) ||
+    !progress.children.every((child) => typeof child === 'string') ||
+    !validAttemptMap(progress.attempts)
+  ) {
+    throw new Error('Corrupted screen sync progress');
+  }
+  return progress as ScreenProgress;
+}
 export type HourlySection = {
   accountId?: string;
   errorCode: null | string;
@@ -43,6 +69,42 @@ export type HourlyProgress = {
   result?: 'failed' | 'partial' | 'succeeded';
   sections: HourlySection[];
 };
+// Same tolerance rule as screenProgress: required fields are the ones every
+// hourly writer has always written; optional collections may be absent.
+export function hourlyProgress(value: Prisma.JsonValue): HourlyProgress {
+  const progress: unknown = value;
+  if (typeof progress !== 'object' || progress === null) {
+    throw new Error('Corrupted hourly sync progress');
+  }
+  const row = progress as Record<string, unknown>;
+  const validChildren =
+    row.children === undefined ||
+    (Array.isArray(row.children) &&
+      row.children.every((child) => {
+        if (typeof child !== 'object' || child === null) return false;
+        const entry = child as Record<string, unknown>;
+        return typeof entry.id === 'string' && typeof entry.section === 'string';
+      }));
+  const validSections =
+    row.sections === undefined ||
+    (Array.isArray(row.sections) &&
+      row.sections.every((section) => {
+        if (typeof section !== 'object' || section === null) return false;
+        const entry = section as Record<string, unknown>;
+        return (
+          typeof entry.section === 'string' &&
+          (typeof entry.errorCode === 'string' || entry.errorCode === null)
+        );
+      }));
+  if (
+    !validSections ||
+    !validAttemptMap(row.attempts) ||
+    !validChildren
+  ) {
+    throw new Error('Corrupted hourly sync progress');
+  }
+  return progress as HourlyProgress;
+}
 type HourlyAttempt = {
   finishedAt: Date | null;
   sections: HourlySection[];
@@ -99,7 +161,7 @@ export class TbankSyncRepository {
         where: { userId_revision_scopeKey: key },
       });
       if (previous?.state === 'running' && previous.processRunId === runId) {
-        const progress = previous.progress as ScreenProgress;
+        const progress = screenProgress(previous.progress);
         const children = progress.children ?? [];
         if (
           !progress.dispatched ||
@@ -115,9 +177,7 @@ export class TbankSyncRepository {
         processRunId: runId,
         state: 'running',
         priority: 0,
-        intent: JSON.parse(
-          JSON.stringify({ scope, now: new Date().toISOString() }),
-        ) as Prisma.InputJsonObject,
+        intent: toJson({ scope, now: new Date().toISOString() }),
         progress: { children: [], dispatched: false },
         errorCode: null,
       };
@@ -136,7 +196,7 @@ export class TbankSyncRepository {
         where: { userId, revision, id },
       });
       if (!work) return null;
-      const progress = work.progress as ScreenProgress;
+      const progress = screenProgress(work.progress);
       const children = await tx.tbankSyncWork.findMany({
         where: { userId, revision, id: { in: progress.children ?? [] } },
         select: { id: true, attemptId: true, state: true, errorCode: true },
@@ -163,7 +223,7 @@ export class TbankSyncRepository {
     return this.prisma.$transaction(async (tx) => {
       const previous = await tx.tbankSyncWork.findFirst({ where: attempt });
       if (!previous) return { count: 0 };
-      const old = previous.progress as ScreenProgress;
+      const old = screenProgress(previous.progress);
       const rows = await tx.tbankSyncWork.findMany({
         where: {
           userId: attempt.userId,
@@ -295,7 +355,7 @@ export class TbankSyncRepository {
     cause: HourlyCause | null,
     dispatchProblems: HourlySection[],
   ) {
-    const progress = work.progress as HourlyProgress;
+    const progress = hourlyProgress(work.progress);
     const children = progress.children ?? [];
     const rows = children.length
       ? await tx.tbankSyncWork.findMany({
@@ -359,7 +419,7 @@ export class TbankSyncRepository {
         select: { progress: true },
       });
       if (!work) return false;
-      const progress = work.progress as HourlyProgress;
+      const progress = hourlyProgress(work.progress);
       const row = await tx.tbankSyncWork.findFirst({
         where: { userId: attempt.userId, id: child.id },
         select: { attemptId: true },
@@ -404,7 +464,7 @@ export class TbankSyncRepository {
       },
     });
     if (!work) return null;
-    const progress = work.progress as HourlyProgress;
+    const progress = hourlyProgress(work.progress);
     let state: 'cancelled' | 'running' | HourlyProgress['result'];
     if (work.state === 'running') state = 'running';
     else if (work.state === 'cancelled') state = 'cancelled';