refactor: isolate operation ranges and intents
This commit is contained in:
@@ -19,6 +19,7 @@ import {
|
|||||||
} from '../../investment-sync-plan';
|
} from '../../investment-sync-plan';
|
||||||
import { OperationErrorCode } from '../domain/operation-error-code';
|
import { OperationErrorCode } from '../domain/operation-error-code';
|
||||||
import {
|
import {
|
||||||
|
confirmedOperationRanges,
|
||||||
operationIntervalsOverlap,
|
operationIntervalsOverlap,
|
||||||
operationsSyncPriority,
|
operationsSyncPriority,
|
||||||
planOperationRead,
|
planOperationRead,
|
||||||
@@ -196,22 +197,7 @@ export class BrokerageOperationsService {
|
|||||||
const plan = planOperationRead(targets, saved?.coverage ?? [], now, {
|
const plan = planOperationRead(targets, saved?.coverage ?? [], now, {
|
||||||
movingEnd,
|
movingEnd,
|
||||||
});
|
});
|
||||||
const confirmed = plan.intervals.flatMap((target) =>
|
const confirmed = confirmedOperationRanges(plan.intervals, saved?.coverage ?? []);
|
||||||
(saved?.coverage ?? [])
|
|
||||||
.filter(
|
|
||||||
(row) =>
|
|
||||||
row.accountId === target.accountId &&
|
|
||||||
(!row.instrumentUid ||
|
|
||||||
row.instrumentUid === target.instrumentUid) &&
|
|
||||||
row.from < target.to &&
|
|
||||||
target.from < row.to,
|
|
||||||
)
|
|
||||||
.map((row) => ({
|
|
||||||
accountId: target.accountId,
|
|
||||||
from: new Date(Math.max(row.from.valueOf(), target.from.valueOf())),
|
|
||||||
to: new Date(Math.min(row.to.valueOf(), target.to.valueOf())),
|
|
||||||
})),
|
|
||||||
);
|
|
||||||
const missing = uncoveredOperationIntervals(
|
const missing = uncoveredOperationIntervals(
|
||||||
plan.intervals,
|
plan.intervals,
|
||||||
confirmed.map((range) => ({
|
confirmed.map((range) => ({
|
||||||
@@ -544,19 +530,7 @@ export class BrokerageOperationsService {
|
|||||||
.sort((a, b) => a.valueOf() - b.valueOf())[0] ??
|
.sort((a, b) => a.valueOf() - b.valueOf())[0] ??
|
||||||
(complete ? saved?.connection.membershipAt : null) ??
|
(complete ? saved?.connection.membershipAt : null) ??
|
||||||
null;
|
null;
|
||||||
const confirmed = plan.intervals.flatMap((target) =>
|
const confirmed = confirmedOperationRanges(plan.intervals, sources);
|
||||||
sources
|
|
||||||
.filter(
|
|
||||||
(row) =>
|
|
||||||
row.accountId === target.accountId &&
|
|
||||||
operationIntervalsOverlap(row, target),
|
|
||||||
)
|
|
||||||
.map((row) => ({
|
|
||||||
accountId: target.accountId,
|
|
||||||
from: new Date(Math.max(row.from.valueOf(), target.from.valueOf())),
|
|
||||||
to: new Date(Math.min(row.to.valueOf(), target.to.valueOf())),
|
|
||||||
})),
|
|
||||||
);
|
|
||||||
const meta = investmentResourceMeta({
|
const meta = investmentResourceMeta({
|
||||||
// eslint-disable-next-line security/detect-object-injection -- The numeric index enumerates the three comparison periods.
|
// eslint-disable-next-line security/detect-object-injection -- The numeric index enumerates the three comparison periods.
|
||||||
resourceKey: `income/${accountId}/${['selected', 'previous', 'yearAgo'][index]}/${period.key}`,
|
resourceKey: `income/${accountId}/${['selected', 'previous', 'yearAgo'][index]}/${period.key}`,
|
||||||
|
|||||||
@@ -1,4 +1,30 @@
|
|||||||
import { operationIntervals, planOperationRead } from './operations-sync-plan';
|
import { confirmedOperationRanges, operationIntervals, planOperationRead } from './operations-sync-plan';
|
||||||
|
|
||||||
|
describe('confirmed operation range projection', () => {
|
||||||
|
it('clips overlaps in source order without merging and respects account and instrument scope', () => {
|
||||||
|
const day = (value: number) => new Date(Date.UTC(2026, 0, value));
|
||||||
|
const target = { accountId: 'a', from: day(2), to: day(5) };
|
||||||
|
const sources = [
|
||||||
|
{ ...target, from: day(3), to: day(6) },
|
||||||
|
{ ...target, from: day(1), to: day(4) },
|
||||||
|
{ ...target, instrumentUid: 'uid' },
|
||||||
|
{ ...target, accountId: 'other' },
|
||||||
|
{ ...target, from: day(5), to: day(6) },
|
||||||
|
{ ...target, from: day(1), to: day(2) },
|
||||||
|
];
|
||||||
|
const before = structuredClone(sources);
|
||||||
|
expect(confirmedOperationRanges([target], sources)).toEqual([
|
||||||
|
{ ...target, from: day(3) },
|
||||||
|
{ ...target, to: day(4) },
|
||||||
|
]);
|
||||||
|
expect(confirmedOperationRanges([{ ...target, instrumentUid: 'uid' }], sources)).toEqual([
|
||||||
|
{ ...target, from: day(3) },
|
||||||
|
{ ...target, to: day(4) },
|
||||||
|
target,
|
||||||
|
]);
|
||||||
|
expect(sources).toEqual(before);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe('operations calendar windows', () => {
|
describe('operations calendar windows', () => {
|
||||||
it('merges overlaps before splitting leap February and keeps precise Moscow-derived boundaries', () => {
|
it('merges overlaps before splitting leap February and keeps precise Moscow-derived boundaries', () => {
|
||||||
|
|||||||
@@ -106,6 +106,24 @@ export function uncoveredOperationIntervals(
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function confirmedOperationRanges(
|
||||||
|
targets: OperationInterval[],
|
||||||
|
coverage: OperationInterval[],
|
||||||
|
): Omit<OperationInterval, 'instrumentUid'>[] {
|
||||||
|
return targets.flatMap((target) => coverage
|
||||||
|
.filter((row) =>
|
||||||
|
row.accountId === target.accountId &&
|
||||||
|
(!row.instrumentUid || row.instrumentUid === target.instrumentUid) &&
|
||||||
|
row.from < target.to && target.from < row.to,
|
||||||
|
)
|
||||||
|
.map((row) => ({
|
||||||
|
accountId: target.accountId,
|
||||||
|
from: new Date(Math.max(row.from.valueOf(), target.from.valueOf())),
|
||||||
|
to: new Date(Math.min(row.to.valueOf(), target.to.valueOf())),
|
||||||
|
})),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
export type OperationCoverage = OperationInterval & {
|
export type OperationCoverage = OperationInterval & {
|
||||||
asOf: Date;
|
asOf: Date;
|
||||||
fetchedAt: Date;
|
fetchedAt: Date;
|
||||||
|
|||||||
+43
-46
@@ -392,52 +392,15 @@ export class BrokerageOperationsRepository {
|
|||||||
const work = await tx.tbankSyncWork.update({
|
const work = await tx.tbankSyncWork.update({
|
||||||
where: { id: previous.id },
|
where: { id: previous.id },
|
||||||
data: {
|
data: {
|
||||||
intent: {
|
intent: extendRunningIntent({
|
||||||
atomic: previousIntent?.atomic ?? publicationAtomic,
|
previous: previousIntent,
|
||||||
atomicIntervals: previousIntent?.atomic
|
publicationAtomic,
|
||||||
? [
|
pending,
|
||||||
...new Set([
|
additions,
|
||||||
...(previousIntent.atomicIntervals ??
|
intervals,
|
||||||
previousIntent.intervals),
|
priority,
|
||||||
...additions.map(intervalKey),
|
previousPriority: previous.priority,
|
||||||
]),
|
}),
|
||||||
]
|
|
||||||
: [],
|
|
||||||
screenIntervals: [
|
|
||||||
...new Set([
|
|
||||||
...(previousIntent?.screenIntervals ??
|
|
||||||
(previous.priority === operationsSyncPriority.screen
|
|
||||||
? (previousIntent?.intervals ?? [])
|
|
||||||
: [])),
|
|
||||||
...(priority === operationsSyncPriority.screen
|
|
||||||
? [...pending, ...additions]
|
|
||||||
.filter((item) =>
|
|
||||||
intervals.some((target) =>
|
|
||||||
operationIntervalsOverlap(item, target),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
.map(intervalKey)
|
|
||||||
: []),
|
|
||||||
]),
|
|
||||||
],
|
|
||||||
intervals: (priority < previous.priority
|
|
||||||
? [
|
|
||||||
...pending.filter((item) =>
|
|
||||||
intervals.some((target) =>
|
|
||||||
operationIntervalsOverlap(item, target),
|
|
||||||
),
|
|
||||||
),
|
|
||||||
...additions,
|
|
||||||
...pending.filter(
|
|
||||||
(item) =>
|
|
||||||
!intervals.some((target) =>
|
|
||||||
operationIntervalsOverlap(item, target),
|
|
||||||
),
|
|
||||||
),
|
|
||||||
]
|
|
||||||
: [...pending, ...additions]
|
|
||||||
).map(intervalKey),
|
|
||||||
},
|
|
||||||
priority: Math.min(priority, previous.priority),
|
priority: Math.min(priority, previous.priority),
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
@@ -919,6 +882,40 @@ 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 {
|
function intervalKey(interval: OperationInterval): string {
|
||||||
// eslint-disable-next-line sonarjs/no-inverted-boolean-check -- Negation also rejects Invalid Date; >= would silently accept NaN.
|
// eslint-disable-next-line sonarjs/no-inverted-boolean-check -- Negation also rejects Invalid Date; >= would silently accept NaN.
|
||||||
if (!(interval.from < interval.to)) {
|
if (!(interval.from < interval.to)) {
|
||||||
|
|||||||
@@ -115,6 +115,40 @@ describe('Request-driven operations synchronization (SQLite)', () => {
|
|||||||
});
|
});
|
||||||
afterAll(async () => prisma.$disconnect());
|
afterAll(async () => prisma.$disconnect());
|
||||||
|
|
||||||
|
it.each([[false, 1], [true, 1], [false, 2], [true, 2]] as const)(
|
||||||
|
'extends legacy intents in stable priority order (atomic=%s, previousPriority=%s)',
|
||||||
|
async (atomic, previousPriority) => {
|
||||||
|
const run = 'legacy-run';
|
||||||
|
const initial = await repository.request('owner', 'revision', [january, february], run, false, new Set(), previousPriority);
|
||||||
|
if (!initial) throw new Error('Missing work');
|
||||||
|
const key = (range: typeof january) => JSON.stringify([
|
||||||
|
range.accountId, range.from.toISOString(), range.to.toISOString(), '',
|
||||||
|
]);
|
||||||
|
const march = { ...january, from: february.to, to: new Date('2026-04-01Z') };
|
||||||
|
await prisma.tbankSyncWork.update({
|
||||||
|
where: { id: initial.work.id },
|
||||||
|
data: { intent: { atomic, intervals: [key(february), key(january)] } },
|
||||||
|
});
|
||||||
|
const extended = await repository.request(
|
||||||
|
'owner', 'revision', [january, march], run, false,
|
||||||
|
new Set([initial.work.attemptId]), 1,
|
||||||
|
);
|
||||||
|
expect(extended?.launch).toBe(false);
|
||||||
|
expect(extended?.work.id).toBe(initial.work.id);
|
||||||
|
expect(extended?.work.priority).toBe(1);
|
||||||
|
expect(extended?.work.intent).toEqual({
|
||||||
|
atomic,
|
||||||
|
atomicIntervals: atomic ? [key(february), key(january), key(march)] : [],
|
||||||
|
screenIntervals: previousPriority === 1
|
||||||
|
? [key(february), key(january), key(march)]
|
||||||
|
: [key(january), key(march)],
|
||||||
|
intervals: previousPriority === 1
|
||||||
|
? [key(february), key(january), key(march)]
|
||||||
|
: [key(january), key(march), key(february)],
|
||||||
|
});
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
it('joins overlapping account and instrument refresh into one atomic publication group', async () => {
|
it('joins overlapping account and instrument refresh into one atomic publication group', async () => {
|
||||||
const gate = deferred<TbankOperationsByCursorResponse>();
|
const gate = deferred<TbankOperationsByCursorResponse>();
|
||||||
upstream.getOperationsByCursor
|
upstream.getOperationsByCursor
|
||||||
|
|||||||
Reference in New Issue
Block a user