refactor: share prisma json codec and check progress shape (WealthLedger/backend#170) #191
@@ -5,11 +5,11 @@ import { isDeepStrictEqual } from 'node:util';
|
|||||||
import type {
|
import type {
|
||||||
TbankInstrumentLookup as MetadataLookupRow,
|
TbankInstrumentLookup as MetadataLookupRow,
|
||||||
TbankInstrument as MetadataRow,
|
TbankInstrument as MetadataRow,
|
||||||
Prisma,
|
|
||||||
} from '../../../generated/prisma/client';
|
} from '../../../generated/prisma/client';
|
||||||
import type { TbankInstrument } from './tbank.client';
|
import type { TbankInstrument } from './tbank.client';
|
||||||
|
|
||||||
import { PrismaService } from '../../prisma/prisma.service';
|
import { PrismaService } from '../../prisma/prisma.service';
|
||||||
|
import { toJson } from '../../prisma/to-json';
|
||||||
import { isJoinableRunningAttempt, type TbankSyncAttempt } from './tbank-access.service';
|
import { isJoinableRunningAttempt, type TbankSyncAttempt } from './tbank-access.service';
|
||||||
import { TbankErrorCode } from './tbank-error-code';
|
import { TbankErrorCode } from './tbank-error-code';
|
||||||
import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim';
|
import { claimSyncWork, lockConnectionGeneration } from './tbank-sync-claim';
|
||||||
@@ -206,9 +206,7 @@ export class InstrumentMetadataRepository {
|
|||||||
const previous = await tx.tbankInstrument.findUnique({
|
const previous = await tx.tbankInstrument.findUnique({
|
||||||
where: { provider_uid: key },
|
where: { provider_uid: key },
|
||||||
});
|
});
|
||||||
const payload = JSON.parse(
|
const payload = toJson(instrument);
|
||||||
JSON.stringify(instrument),
|
|
||||||
) as Prisma.InputJsonObject;
|
|
||||||
const data = {
|
const data = {
|
||||||
metadata: payload,
|
metadata: payload,
|
||||||
figi:
|
figi:
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import {
|
|||||||
type TbankSyncAttempt,
|
type TbankSyncAttempt,
|
||||||
} from '../../../../integrations/tbank/tbank.api';
|
} from '../../../../integrations/tbank/tbank.api';
|
||||||
import { PrismaService } from '../../../../prisma/prisma.service';
|
import { PrismaService } from '../../../../prisma/prisma.service';
|
||||||
|
import { toJson } from '../../../../prisma/to-json';
|
||||||
import {
|
import {
|
||||||
type CandleRange,
|
type CandleRange,
|
||||||
confirmedCandleCoverage,
|
confirmedCandleCoverage,
|
||||||
@@ -118,7 +119,7 @@ export class CandleRepository {
|
|||||||
processRunId: runId,
|
processRunId: runId,
|
||||||
state: 'running',
|
state: 'running',
|
||||||
priority,
|
priority,
|
||||||
intent: JSON.parse(JSON.stringify(intent)) as Prisma.InputJsonObject,
|
intent: toJson(intent),
|
||||||
progress: {},
|
progress: {},
|
||||||
errorCode: null,
|
errorCode: null,
|
||||||
};
|
};
|
||||||
@@ -176,9 +177,7 @@ export class CandleRepository {
|
|||||||
sourceKey: 'candles',
|
sourceKey: 'candles',
|
||||||
itemKey: key,
|
itemKey: key,
|
||||||
};
|
};
|
||||||
const payload = JSON.parse(
|
const payload = toJson({ range, candles: rows, fetchedAt });
|
||||||
JSON.stringify({ range, candles: rows, fetchedAt }),
|
|
||||||
) as Prisma.InputJsonObject;
|
|
||||||
await tx.tbankSyncStage.upsert({
|
await tx.tbankSyncStage.upsert({
|
||||||
where: { workId_attemptId_sourceKey_itemKey: identity },
|
where: { workId_attemptId_sourceKey_itemKey: identity },
|
||||||
create: { ...identity, payload },
|
create: { ...identity, payload },
|
||||||
|
|||||||
+20
-27
@@ -2,12 +2,11 @@ import { Injectable } from '@nestjs/common';
|
|||||||
import { randomUUID } from 'node:crypto';
|
import { randomUUID } from 'node:crypto';
|
||||||
import { isDeepStrictEqual } from 'node:util';
|
import { isDeepStrictEqual } from 'node:util';
|
||||||
|
|
||||||
import type { Prisma } from '../../../../../generated/prisma/client';
|
|
||||||
|
|
||||||
import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
|
import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
|
||||||
import { claimSyncWork, lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim';
|
import { claimSyncWork, lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim';
|
||||||
import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
|
import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
|
||||||
import { PrismaService } from '../../../../prisma/prisma.service';
|
import { PrismaService } from '../../../../prisma/prisma.service';
|
||||||
|
import { toJson } from '../../../../prisma/to-json';
|
||||||
|
|
||||||
const UNCHANGED_REVISION_INCREMENT = 0;
|
const UNCHANGED_REVISION_INCREMENT = 0;
|
||||||
const REVISION_INCREMENT = 1;
|
const REVISION_INCREMENT = 1;
|
||||||
@@ -172,9 +171,7 @@ export class CatalogDiscoveryRepository {
|
|||||||
progress: {},
|
progress: {},
|
||||||
});
|
});
|
||||||
if (!claimed) return false;
|
if (!claimed) return false;
|
||||||
const payload = JSON.parse(
|
const payload = toJson(value);
|
||||||
JSON.stringify(value),
|
|
||||||
) as Prisma.InputJsonObject;
|
|
||||||
const identity = { provider, uid: input.uid };
|
const identity = { provider, uid: input.uid };
|
||||||
await tx.tbankInstrument.upsert({
|
await tx.tbankInstrument.upsert({
|
||||||
where: { provider_uid: identity },
|
where: { provider_uid: identity },
|
||||||
@@ -197,12 +194,10 @@ export class CatalogDiscoveryRepository {
|
|||||||
const previous = await tx.tbankInstrument.findUniqueOrThrow({
|
const previous = await tx.tbankInstrument.findUniqueOrThrow({
|
||||||
where: { provider_uid: identity },
|
where: { provider_uid: identity },
|
||||||
});
|
});
|
||||||
const metadata = JSON.parse(
|
const metadata = toJson({
|
||||||
JSON.stringify({
|
...instrument,
|
||||||
...instrument,
|
instrumentType: input.instrumentType,
|
||||||
instrumentType: input.instrumentType,
|
});
|
||||||
}),
|
|
||||||
) as Prisma.InputJsonObject;
|
|
||||||
await tx.tbankInstrument.update({
|
await tx.tbankInstrument.update({
|
||||||
where: { provider_uid: identity },
|
where: { provider_uid: identity },
|
||||||
data: {
|
data: {
|
||||||
@@ -341,22 +336,20 @@ export class CatalogDiscoveryRepository {
|
|||||||
for (const instrument of instruments) {
|
for (const instrument of instruments) {
|
||||||
// Discovery can create a minimal identity, but cannot overwrite verified metadata.
|
// Discovery can create a minimal identity, but cannot overwrite verified metadata.
|
||||||
const identity = { provider, uid: instrument.uid };
|
const identity = { provider, uid: instrument.uid };
|
||||||
const payload = JSON.parse(
|
const payload = toJson({
|
||||||
JSON.stringify({
|
uid: instrument.uid,
|
||||||
uid: instrument.uid,
|
figi: instrument.figi,
|
||||||
figi: instrument.figi,
|
ticker: instrument.ticker,
|
||||||
ticker: instrument.ticker,
|
name: instrument.name,
|
||||||
name: instrument.name,
|
isin: instrument.isin,
|
||||||
isin: instrument.isin,
|
classCode: instrument.classCode,
|
||||||
classCode: instrument.classCode,
|
instrumentKind: instrument.instrumentKind,
|
||||||
instrumentKind: instrument.instrumentKind,
|
apiTradeAvailableFlag: instrument.apiTradeAvailableFlag,
|
||||||
apiTradeAvailableFlag: instrument.apiTradeAvailableFlag,
|
currency: instrument.currency,
|
||||||
currency: instrument.currency,
|
brand: instrument.brand
|
||||||
brand: instrument.brand
|
? { logoName: instrument.brand.logoName }
|
||||||
? { logoName: instrument.brand.logoName }
|
: undefined,
|
||||||
: undefined,
|
});
|
||||||
}),
|
|
||||||
) as Prisma.InputJsonObject;
|
|
||||||
await tx.tbankInstrument.upsert({
|
await tx.tbankInstrument.upsert({
|
||||||
where: { provider_uid: identity },
|
where: { provider_uid: identity },
|
||||||
update: {},
|
update: {},
|
||||||
|
|||||||
+2
-3
@@ -12,6 +12,7 @@ import {
|
|||||||
type TbankOperationItem, type TbankOperationsByCursorResponse,
|
type TbankOperationItem, type TbankOperationsByCursorResponse,
|
||||||
} from '../../../../integrations/tbank/tbank.api';
|
} from '../../../../integrations/tbank/tbank.api';
|
||||||
import { PrismaService } from '../../../../prisma/prisma.service';
|
import { PrismaService } from '../../../../prisma/prisma.service';
|
||||||
|
import { toJson } from '../../../../prisma/to-json';
|
||||||
import {
|
import {
|
||||||
decodeInterval,
|
decodeInterval,
|
||||||
extendRunningIntent,
|
extendRunningIntent,
|
||||||
@@ -469,9 +470,7 @@ export class BrokerageOperationsRepository {
|
|||||||
) {
|
) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
const payload = JSON.parse(
|
const payload = toJson({ ...operation, cursor: undefined });
|
||||||
JSON.stringify({ ...operation, cursor: undefined }),
|
|
||||||
) as Prisma.InputJsonObject;
|
|
||||||
// Provider pagination position is not part of a financial row.
|
// Provider pagination position is not part of a financial row.
|
||||||
const identity = { ...stage, itemKey: operation.id };
|
const identity = { ...stage, itemKey: operation.id };
|
||||||
const previous = await tx.tbankSyncStage.findUnique({
|
const previous = await tx.tbankSyncStage.findUnique({
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import { isDeepStrictEqual } from 'node:util';
|
|||||||
import type { Prisma } from '../../../../../generated/prisma/client';
|
import type { Prisma } from '../../../../../generated/prisma/client';
|
||||||
|
|
||||||
import { PrismaService } from '../../../../prisma/prisma.service';
|
import { PrismaService } from '../../../../prisma/prisma.service';
|
||||||
|
import { toJson } from '../../../../prisma/to-json';
|
||||||
|
|
||||||
const FIRST_BATCH_OFFSET = 0;
|
const FIRST_BATCH_OFFSET = 0;
|
||||||
|
|
||||||
@@ -142,13 +143,11 @@ export class ForecastRepository {
|
|||||||
) {
|
) {
|
||||||
return fallback;
|
return fallback;
|
||||||
}
|
}
|
||||||
const revisions = JSON.parse(
|
const revisions = toJson({
|
||||||
JSON.stringify({
|
portfolioProjection: input.portfolioProjection,
|
||||||
portfolioProjection: input.portfolioProjection,
|
portfolios: input.portfolios,
|
||||||
portfolios: input.portfolios,
|
sources: input.sources,
|
||||||
sources: input.sources,
|
});
|
||||||
}),
|
|
||||||
) as Prisma.InputJsonObject;
|
|
||||||
const timestamps = input.timestamps as Prisma.InputJsonArray;
|
const timestamps = input.timestamps as Prisma.InputJsonArray;
|
||||||
if (
|
if (
|
||||||
fallback?.scopeKey === input.scopeKey &&
|
fallback?.scopeKey === input.scopeKey &&
|
||||||
@@ -159,9 +158,7 @@ export class ForecastRepository {
|
|||||||
}
|
}
|
||||||
const data = {
|
const data = {
|
||||||
membershipRevision: input.membershipRevision,
|
membershipRevision: input.membershipRevision,
|
||||||
payload: JSON.parse(
|
payload: toJson(input.candidate),
|
||||||
JSON.stringify(input.candidate),
|
|
||||||
) as Prisma.InputJsonObject,
|
|
||||||
dependencyRevisions: revisions,
|
dependencyRevisions: revisions,
|
||||||
sourceTimestamps: timestamps,
|
sourceTimestamps: timestamps,
|
||||||
calculatedAt: new Date(),
|
calculatedAt: new Date(),
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'
|
|||||||
import { lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim';
|
import { lockConnectionGeneration } from '../../../../integrations/tbank/tbank-sync-claim';
|
||||||
import { isJoinableRunningAttempt, tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api';
|
import { isJoinableRunningAttempt, tbankSourceNeedsRefresh, tbankSourceTtlMs } from '../../../../integrations/tbank/tbank.api';
|
||||||
import { PrismaService } from '../../../../prisma/prisma.service';
|
import { PrismaService } from '../../../../prisma/prisma.service';
|
||||||
|
import { toJson } from '../../../../prisma/to-json';
|
||||||
|
|
||||||
const EXPECTED_UPDATED_ACCOUNTS = 1;
|
const EXPECTED_UPDATED_ACCOUNTS = 1;
|
||||||
|
|
||||||
@@ -128,7 +129,7 @@ export class PortfolioRepository {
|
|||||||
accountId,
|
accountId,
|
||||||
};
|
};
|
||||||
const data = {
|
const data = {
|
||||||
payload: JSON.parse(JSON.stringify(snapshot)) as Prisma.InputJsonObject,
|
payload: toJson(snapshot),
|
||||||
fetchedAt: new Date(snapshot.lastSuccessfulAt),
|
fetchedAt: new Date(snapshot.lastSuccessfulAt),
|
||||||
};
|
};
|
||||||
await tx.tbankPortfolio.upsert({
|
await tx.tbankPortfolio.upsert({
|
||||||
|
|||||||
@@ -2,7 +2,11 @@ jest.mock('../../../prisma/prisma.service', () => ({
|
|||||||
PrismaService: class {},
|
PrismaService: class {},
|
||||||
}));
|
}));
|
||||||
|
|
||||||
import { screenIntent } from './tbank-sync.repository';
|
import {
|
||||||
|
hourlyProgress,
|
||||||
|
screenIntent,
|
||||||
|
screenProgress,
|
||||||
|
} from './tbank-sync.repository';
|
||||||
|
|
||||||
describe('screen sync intent', () => {
|
describe('screen sync intent', () => {
|
||||||
it('restores a persisted intent and rejects a corrupted one', () => {
|
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');
|
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',
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import type { InvestmentScreenDto } from './tbank-sync.dto';
|
|||||||
|
|
||||||
import { claimSyncWork, lockConnectionGeneration } from '../../../integrations/tbank/tbank-sync-claim';
|
import { claimSyncWork, lockConnectionGeneration } from '../../../integrations/tbank/tbank-sync-claim';
|
||||||
import { PrismaService } from '../../../prisma/prisma.service';
|
import { PrismaService } from '../../../prisma/prisma.service';
|
||||||
|
import { toJson } from '../../../prisma/to-json';
|
||||||
export type ScreenIntent = { now: string; scope: InvestmentScreenDto };
|
export type ScreenIntent = { now: string; scope: InvestmentScreenDto };
|
||||||
|
|
||||||
// Json columns are written only by this repository; a shape check turns corruption into an explicit 500.
|
// 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[];
|
children: string[];
|
||||||
dispatched: boolean;
|
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 = {
|
export type HourlySection = {
|
||||||
accountId?: string;
|
accountId?: string;
|
||||||
errorCode: null | string;
|
errorCode: null | string;
|
||||||
@@ -44,6 +70,42 @@ export type HourlyProgress = {
|
|||||||
result?: 'failed' | 'partial' | 'succeeded';
|
result?: 'failed' | 'partial' | 'succeeded';
|
||||||
sections: HourlySection[];
|
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 = {
|
type HourlyAttempt = {
|
||||||
finishedAt: Date | null;
|
finishedAt: Date | null;
|
||||||
sections: HourlySection[];
|
sections: HourlySection[];
|
||||||
@@ -97,7 +159,7 @@ export class TbankSyncRepository {
|
|||||||
where: { userId_revision_scopeKey: key },
|
where: { userId_revision_scopeKey: key },
|
||||||
});
|
});
|
||||||
if (previous?.state === 'running' && previous.processRunId === runId) {
|
if (previous?.state === 'running' && previous.processRunId === runId) {
|
||||||
const progress = previous.progress as ScreenProgress;
|
const progress = screenProgress(previous.progress);
|
||||||
const children = progress.children ?? [];
|
const children = progress.children ?? [];
|
||||||
if (
|
if (
|
||||||
!progress.dispatched ||
|
!progress.dispatched ||
|
||||||
@@ -113,9 +175,7 @@ export class TbankSyncRepository {
|
|||||||
processRunId: runId,
|
processRunId: runId,
|
||||||
state: 'running',
|
state: 'running',
|
||||||
priority: 0,
|
priority: 0,
|
||||||
intent: JSON.parse(
|
intent: toJson({ scope, now: new Date().toISOString() }),
|
||||||
JSON.stringify({ scope, now: new Date().toISOString() }),
|
|
||||||
) as Prisma.InputJsonObject,
|
|
||||||
progress: { children: [], dispatched: false },
|
progress: { children: [], dispatched: false },
|
||||||
errorCode: null,
|
errorCode: null,
|
||||||
};
|
};
|
||||||
@@ -134,7 +194,7 @@ export class TbankSyncRepository {
|
|||||||
where: { userId, revision, id },
|
where: { userId, revision, id },
|
||||||
});
|
});
|
||||||
if (!work) return null;
|
if (!work) return null;
|
||||||
const progress = work.progress as ScreenProgress;
|
const progress = screenProgress(work.progress);
|
||||||
const children = await tx.tbankSyncWork.findMany({
|
const children = await tx.tbankSyncWork.findMany({
|
||||||
where: { userId, revision, id: { in: progress.children ?? [] } },
|
where: { userId, revision, id: { in: progress.children ?? [] } },
|
||||||
select: { id: true, attemptId: true, state: true, errorCode: true },
|
select: { id: true, attemptId: true, state: true, errorCode: true },
|
||||||
@@ -161,7 +221,7 @@ export class TbankSyncRepository {
|
|||||||
return this.prisma.$transaction(async (tx) => {
|
return this.prisma.$transaction(async (tx) => {
|
||||||
const previous = await tx.tbankSyncWork.findFirst({ where: attempt });
|
const previous = await tx.tbankSyncWork.findFirst({ where: attempt });
|
||||||
if (!previous) return { count: 0 };
|
if (!previous) return { count: 0 };
|
||||||
const old = previous.progress as ScreenProgress;
|
const old = screenProgress(previous.progress);
|
||||||
const rows = await tx.tbankSyncWork.findMany({
|
const rows = await tx.tbankSyncWork.findMany({
|
||||||
where: {
|
where: {
|
||||||
userId: attempt.userId,
|
userId: attempt.userId,
|
||||||
@@ -287,7 +347,7 @@ export class TbankSyncRepository {
|
|||||||
cause: HourlyCause | null,
|
cause: HourlyCause | null,
|
||||||
dispatchProblems: HourlySection[],
|
dispatchProblems: HourlySection[],
|
||||||
) {
|
) {
|
||||||
const progress = work.progress as HourlyProgress;
|
const progress = hourlyProgress(work.progress);
|
||||||
const children = progress.children ?? [];
|
const children = progress.children ?? [];
|
||||||
const rows = children.length
|
const rows = children.length
|
||||||
? await tx.tbankSyncWork.findMany({
|
? await tx.tbankSyncWork.findMany({
|
||||||
@@ -351,7 +411,7 @@ export class TbankSyncRepository {
|
|||||||
select: { progress: true },
|
select: { progress: true },
|
||||||
});
|
});
|
||||||
if (!work) return false;
|
if (!work) return false;
|
||||||
const progress = work.progress as HourlyProgress;
|
const progress = hourlyProgress(work.progress);
|
||||||
const row = await tx.tbankSyncWork.findFirst({
|
const row = await tx.tbankSyncWork.findFirst({
|
||||||
where: { userId: attempt.userId, id: child.id },
|
where: { userId: attempt.userId, id: child.id },
|
||||||
select: { attemptId: true },
|
select: { attemptId: true },
|
||||||
@@ -393,7 +453,7 @@ export class TbankSyncRepository {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
if (!work) return null;
|
if (!work) return null;
|
||||||
const progress = work.progress as HourlyProgress;
|
const progress = hourlyProgress(work.progress);
|
||||||
let state: 'cancelled' | 'running' | HourlyProgress['result'];
|
let state: 'cancelled' | 'running' | HourlyProgress['result'];
|
||||||
if (work.state === 'running') state = 'running';
|
if (work.state === 'running') state = 'running';
|
||||||
else if (work.state === 'cancelled') state = 'cancelled';
|
else if (work.state === 'cancelled') state = 'cancelled';
|
||||||
|
|||||||
@@ -0,0 +1,14 @@
|
|||||||
|
import { toJson } from './to-json';
|
||||||
|
|
||||||
|
it('produces the same bytes as the inline JSON round-trip', () => {
|
||||||
|
const value = {
|
||||||
|
a: 'x',
|
||||||
|
b: undefined,
|
||||||
|
c: new Date('2026-11-04T00:00:00.000Z'),
|
||||||
|
d: [{ e: 1 }],
|
||||||
|
};
|
||||||
|
expect(toJson(value)).toEqual(
|
||||||
|
JSON.parse(JSON.stringify(value)) as object,
|
||||||
|
);
|
||||||
|
expect(toJson(value)).toEqual({ a: 'x', c: '2026-11-04T00:00:00.000Z', d: [{ e: 1 }] });
|
||||||
|
});
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
import type { Prisma } from '../../generated/prisma/client';
|
||||||
|
|
||||||
|
// Single owner for the Prisma Json write codec (backend#170): a JSON
|
||||||
|
// round-trip drops undefined values and Date instances become ISO strings,
|
||||||
|
// which is exactly what the Json columns expect. Same operations as the
|
||||||
|
// inline copies it replaced, so written bytes are identical.
|
||||||
|
export function toJson<T extends object>(value: T): Prisma.InputJsonObject {
|
||||||
|
return JSON.parse(JSON.stringify(value)) as Prisma.InputJsonObject;
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user