chore: strengthen backend eslint profile
This commit is contained in:
+36
-27
@@ -1,22 +1,23 @@
|
||||
import { accountHistory } from '../../investment-sync-plan';
|
||||
import { Injectable, NotFoundException } from '@nestjs/common';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
|
||||
|
||||
import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
|
||||
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
|
||||
import {
|
||||
TbankAccessService,
|
||||
TbankClient,
|
||||
} from '../../../../integrations/tbank/tbank.api';
|
||||
import { accountHistory } from '../../investment-sync-plan';
|
||||
import {
|
||||
type OperationInterval,
|
||||
operationIntervals,
|
||||
planOperationRead,
|
||||
uncoveredOperationIntervals,
|
||||
} from '../domain/operations-sync-plan';
|
||||
import {
|
||||
BrokerageOperationsRepository,
|
||||
type OperationAttempt,
|
||||
} from '../infrastructure/brokerage-operations.repository';
|
||||
import {
|
||||
operationIntervals,
|
||||
planOperationRead,
|
||||
type OperationInterval,
|
||||
uncoveredOperationIntervals,
|
||||
} from '../domain/operations-sync-plan';
|
||||
|
||||
const operationsPageSize = 1000;
|
||||
const priorities = { screen: 1, history: 2 } as const;
|
||||
@@ -29,6 +30,7 @@ export class BrokerageOperationsSyncService {
|
||||
string,
|
||||
{ attempt: OperationAttempt; priority: number }
|
||||
>();
|
||||
|
||||
private draining = false;
|
||||
// Only short DB reservations/publications are serialized, never RPCs.
|
||||
private reservations: Promise<unknown> = Promise.resolve();
|
||||
@@ -44,10 +46,10 @@ export class BrokerageOperationsSyncService {
|
||||
userId: string,
|
||||
intervals: OperationInterval[],
|
||||
options: {
|
||||
movingEnd?: boolean;
|
||||
force?: boolean;
|
||||
observe?: boolean;
|
||||
history?: boolean;
|
||||
movingEnd?: boolean;
|
||||
observe?: boolean;
|
||||
revision?: string;
|
||||
} = {},
|
||||
) {
|
||||
@@ -55,8 +57,9 @@ export class BrokerageOperationsSyncService {
|
||||
if (
|
||||
!connection.membershipAt ||
|
||||
(options.revision && options.revision !== connection.revision)
|
||||
)
|
||||
) {
|
||||
return { plan: null, handle: null };
|
||||
}
|
||||
if (
|
||||
intervals.some(
|
||||
(interval) =>
|
||||
@@ -64,8 +67,9 @@ export class BrokerageOperationsSyncService {
|
||||
(account) => account.accountId === interval.accountId,
|
||||
),
|
||||
)
|
||||
)
|
||||
) {
|
||||
throw new NotFoundException();
|
||||
}
|
||||
const coverage = await this.repository.coverage(
|
||||
userId,
|
||||
connection.revision,
|
||||
@@ -76,24 +80,24 @@ export class BrokerageOperationsSyncService {
|
||||
...intervals.map((interval) => interval.to.valueOf()),
|
||||
);
|
||||
const targets = intervals.map((interval) =>
|
||||
options.movingEnd && interval.to.valueOf() === movingTo
|
||||
? { ...interval, to: now }
|
||||
: interval,
|
||||
options.movingEnd && interval.to.valueOf() === movingTo ?
|
||||
{ ...interval, to: now } :
|
||||
interval,
|
||||
);
|
||||
const plan = planOperationRead(targets, coverage, now, options);
|
||||
const complete = plan.intervals.every(
|
||||
(interval) => interval.availability === 'complete',
|
||||
);
|
||||
const handle =
|
||||
!options.observe && plan.load.length
|
||||
? await this.request(
|
||||
!options.observe && plan.load.length ?
|
||||
await this.request(
|
||||
userId,
|
||||
plan.load,
|
||||
complete || Boolean(options.force),
|
||||
'screen',
|
||||
connection.revision,
|
||||
)
|
||||
: null;
|
||||
) :
|
||||
null;
|
||||
if (!options.observe && !options.force && options.history) {
|
||||
const accountIds = new Set(
|
||||
intervals.map((interval) => interval.accountId),
|
||||
@@ -108,7 +112,7 @@ export class BrokerageOperationsSyncService {
|
||||
movingEnd: true,
|
||||
}).load;
|
||||
const remaining = uncoveredOperationIntervals(missing, targets);
|
||||
if (remaining.length)
|
||||
if (remaining.length) {
|
||||
await this.request(
|
||||
userId,
|
||||
remaining,
|
||||
@@ -116,6 +120,7 @@ export class BrokerageOperationsSyncService {
|
||||
'history',
|
||||
connection.revision,
|
||||
);
|
||||
}
|
||||
}
|
||||
return { plan, handle };
|
||||
}
|
||||
@@ -126,7 +131,7 @@ export class BrokerageOperationsSyncService {
|
||||
atomic: boolean,
|
||||
priority: keyof typeof priorities = 'screen',
|
||||
revision?: string,
|
||||
): Promise<string | null> {
|
||||
): Promise<null | string> {
|
||||
const connection = await this.access.local(userId);
|
||||
if (
|
||||
connection.membershipAt &&
|
||||
@@ -136,8 +141,9 @@ export class BrokerageOperationsSyncService {
|
||||
(account) => account.accountId === interval.accountId,
|
||||
),
|
||||
)
|
||||
)
|
||||
) {
|
||||
throw new NotFoundException();
|
||||
}
|
||||
if (revision && revision !== connection.revision) return null;
|
||||
const cutoff = new Date();
|
||||
const bounded = operationIntervals(intervals)
|
||||
@@ -155,6 +161,7 @@ export class BrokerageOperationsSyncService {
|
||||
this.processRunId,
|
||||
atomic,
|
||||
this.runningAttempts,
|
||||
// eslint-disable-next-line security/detect-object-injection -- The priority belongs to the internal priority union.
|
||||
priorities[priority],
|
||||
);
|
||||
if (!requested) return null;
|
||||
@@ -258,17 +265,18 @@ export class BrokerageOperationsSyncService {
|
||||
Math.ceil(interval.to.valueOf() / MILLISECONDS_PER_SECOND) + 1,
|
||||
),
|
||||
},
|
||||
...(interval.instrumentUid
|
||||
? { instrumentId: interval.instrumentUid }
|
||||
: {}),
|
||||
...(interval.instrumentUid ?
|
||||
{ instrumentId: interval.instrumentUid } :
|
||||
{}),
|
||||
cursor,
|
||||
limit: operationsPageSize,
|
||||
state: 'OPERATION_STATE_EXECUTED',
|
||||
});
|
||||
if (
|
||||
!(await this.repository.stagePage(attempt, interval, cursor, response))
|
||||
)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (!plan.atomic && !response.hasNext) {
|
||||
if (
|
||||
!(await this.repository.publish(
|
||||
@@ -277,8 +285,9 @@ export class BrokerageOperationsSyncService {
|
||||
new Date(),
|
||||
true,
|
||||
))
|
||||
)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
// One page per turn: newly requested screen work can overtake backfill.
|
||||
// TbankClient alone owns the bounded RPC queue and retries.
|
||||
|
||||
Reference in New Issue
Block a user