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

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

@@ -8,6 +8,7 @@ import type { InvestmentScreenDto } from './tbank-sync.dto';
 
 import { claimSyncWork, lockConnectionGeneration } from '../../../integrations/tbank/tbank-sync-claim';
 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.
@@ -23,6 +24,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;
@@ -44,6 +70,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[];
@@ -97,7 +159,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 ||
@@ -113,9 +175,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,
       };
@@ -134,7 +194,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 },
@@ -161,7 +221,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,
@@ -287,7 +347,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({
@@ -351,7 +411,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 },
@@ -393,7 +453,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';