refactor!: simplify runtime coordination #7
Keep financial ownership, atomic publication and refresh CAS. Failed refresh leaves cookies untouched; logout still clears them. Remove full-history candle prefetch and special metadata retries. Use native Pino request context instead of parallel middleware. BREAKING CHANGE: remove investment stateVersion, queued sync state, and tbank-sync status progress/resources. Consumers must observe resources through sync=observe with the updated frontend contract.
This commit is contained in:
@@ -2,7 +2,6 @@ import { Injectable } from '@nestjs/common';
|
||||
|
||||
import type { InvestmentResourceMetaDto } from '../../investment-resource.dto';
|
||||
|
||||
import { tbankSourceNeedsRefresh } from '../../../../integrations/tbank/tbank.api';
|
||||
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
||||
import {
|
||||
ForecastRepository,
|
||||
@@ -10,7 +9,6 @@ import {
|
||||
} from '../infrastructure/forecast.repository';
|
||||
|
||||
const MISSING_DATA_REVISION = 0;
|
||||
const STALE_FRESHNESS_MS = 0;
|
||||
|
||||
@Injectable()
|
||||
export class ForecastService {
|
||||
@@ -73,23 +71,8 @@ export class ForecastService {
|
||||
dependencies
|
||||
.map((dependency) => dependency.sync)
|
||||
.find((sync) => sync.state === 'failed');
|
||||
const payload = stored ? (stored.payload as T) : null;
|
||||
const data =
|
||||
payload &&
|
||||
'positionIsStale' in payload &&
|
||||
'positionAt' in payload &&
|
||||
typeof payload.positionAt === 'string'
|
||||
? {
|
||||
...payload,
|
||||
positionIsStale: tbankSourceNeedsRefresh(
|
||||
'portfolio',
|
||||
new Date(payload.positionAt),
|
||||
new Date(Date.now()),
|
||||
),
|
||||
}
|
||||
: payload;
|
||||
return {
|
||||
data,
|
||||
data: stored ? (stored.payload as T) : null,
|
||||
meta: investmentResourceMeta({
|
||||
resourceKey: input.scopeKey,
|
||||
dataRevision: String(stored?.dataRevision ?? MISSING_DATA_REVISION),
|
||||
@@ -101,10 +84,7 @@ export class ForecastService {
|
||||
? 'partial'
|
||||
: 'missing',
|
||||
fetchedAt: first ? new Date(first) : null,
|
||||
freshForMs:
|
||||
data && 'positionIsStale' in data && data.positionIsStale === true
|
||||
? STALE_FRESHNESS_MS
|
||||
: Number.POSITIVE_INFINITY,
|
||||
freshForMs: Number.POSITIVE_INFINITY,
|
||||
work: work?.id ? { ...work, id: work.id } : null,
|
||||
dependencies,
|
||||
}),
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import { ServiceUnavailableException } from '@nestjs/common';
|
||||
import { setImmediate as immediate } from 'node:timers/promises';
|
||||
jest.mock('../../../../prisma/prisma.service', () => ({
|
||||
PrismaService: class {},
|
||||
@@ -23,12 +22,10 @@ describe('persistent portfolio projection', () => {
|
||||
service = new PortfolioService(
|
||||
access as never,
|
||||
tbank as never,
|
||||
{ set: jest.fn() } as never,
|
||||
metadata as never,
|
||||
repository as never,
|
||||
{ request: jest.fn().mockResolvedValue(null) } as never,
|
||||
{ event: jest.fn() } as never,
|
||||
{ prefetch: jest.fn().mockResolvedValue(undefined) } as never,
|
||||
);
|
||||
access.local.mockResolvedValue({
|
||||
revision: 'private',
|
||||
@@ -221,15 +218,4 @@ describe('persistent portfolio projection', () => {
|
||||
expect(complete.data?.isStale).toBe(true);
|
||||
expect(tbank.getPortfolio).not.toHaveBeenCalled();
|
||||
});
|
||||
it('keeps pending data unavailable to internal financial consumers', async () => {
|
||||
repository.read.mockResolvedValue({
|
||||
membershipAt: null,
|
||||
membershipRevision: 0,
|
||||
accounts: [],
|
||||
work: [],
|
||||
});
|
||||
await expect(service.getAll('user')).rejects.toBeInstanceOf(
|
||||
ServiceUnavailableException,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,11 +1,9 @@
|
||||
import {
|
||||
Injectable,
|
||||
NotFoundException,
|
||||
ServiceUnavailableException,
|
||||
} from '@nestjs/common';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
|
||||
import { RequestContextService } from '../../../../common/logging/request-context.service';
|
||||
import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
|
||||
import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
|
||||
import {
|
||||
@@ -19,7 +17,6 @@ import {
|
||||
type TbankPortfolio,
|
||||
tbankSourceTtlMs,
|
||||
} from '../../../../integrations/tbank/tbank.api';
|
||||
import { CandleService } from '../../instruments/instruments.api';
|
||||
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
||||
import { portfolioStructure } from '../domain/portfolio-structure';
|
||||
import { sumMoneyByCurrency } from '../domain/portfolio-totals';
|
||||
@@ -38,23 +35,19 @@ const membershipFreshForMs = tbankSourceTtlMs.membership;
|
||||
@Injectable()
|
||||
export class PortfolioService {
|
||||
private readonly processRunId = randomUUID();
|
||||
private readonly active = new Set<string>();
|
||||
private readonly pending = new Map<string, Promise<void>>();
|
||||
private reservations: Promise<unknown> = Promise.resolve();
|
||||
|
||||
constructor(
|
||||
private readonly access: TbankAccessService,
|
||||
private readonly tbank: TbankClient,
|
||||
private readonly requestContext: RequestContextService,
|
||||
private readonly metadata: InstrumentMetadataService,
|
||||
private readonly repository: PortfolioRepository,
|
||||
private readonly membership: TbankMembershipService,
|
||||
private readonly logger: StructuredLogger,
|
||||
private readonly candles: CandleService,
|
||||
) {}
|
||||
|
||||
async saved(userId: string, accountId: string, observe = false) {
|
||||
this.requestContext.set({ userId, brokerageAccountId: accountId });
|
||||
const connection = await this.access.local(userId, accountId);
|
||||
if (!observe) {
|
||||
const discovery = await this.membership.request(
|
||||
@@ -123,28 +116,6 @@ export class PortfolioService {
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!observe) {
|
||||
void this.candles
|
||||
.prefetch(
|
||||
userId,
|
||||
connection.revision,
|
||||
(saved?.accounts ?? [])
|
||||
.filter(
|
||||
(account) =>
|
||||
accountId === 'all' || account.accountId === accountId,
|
||||
)
|
||||
.flatMap(
|
||||
(account) =>
|
||||
(
|
||||
account.portfolio?.payload as unknown as
|
||||
PortfolioDto | undefined
|
||||
)?.positions ?? [],
|
||||
),
|
||||
)
|
||||
.catch(() =>
|
||||
this.logger.event('warn', 'portfolio.candle-prefetch-unavailable'),
|
||||
);
|
||||
}
|
||||
const membershipMeta = investmentResourceMeta({
|
||||
resourceKey: 'accounts/membership',
|
||||
dataRevision: String(saved?.membershipRevision ?? MISSING_DATA_REVISION),
|
||||
@@ -319,7 +290,7 @@ export class PortfolioService {
|
||||
revision,
|
||||
accountId,
|
||||
this.processRunId,
|
||||
this.active,
|
||||
this.pending,
|
||||
force,
|
||||
);
|
||||
if (!requested) return null;
|
||||
@@ -330,7 +301,6 @@ export class PortfolioService {
|
||||
};
|
||||
}
|
||||
const { id, attemptId } = requested.work;
|
||||
this.active.add(attemptId);
|
||||
const done = new Promise<void>((resolve) => setImmediate(resolve))
|
||||
.then(async () => {
|
||||
const attempt = { id, attemptId, userId, revision };
|
||||
@@ -354,13 +324,6 @@ export class PortfolioService {
|
||||
for (const position of snapshot.positions) {
|
||||
await this.metadata.request(userId, revision, position);
|
||||
}
|
||||
if (!force) {
|
||||
await this.candles.prefetch(
|
||||
userId,
|
||||
revision,
|
||||
snapshot.positions,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
if (
|
||||
@@ -377,7 +340,6 @@ export class PortfolioService {
|
||||
}),
|
||||
)
|
||||
.finally(() => {
|
||||
this.active.delete(attemptId);
|
||||
this.pending.delete(attemptId);
|
||||
});
|
||||
this.pending.set(attemptId, done);
|
||||
@@ -387,18 +349,6 @@ export class PortfolioService {
|
||||
return reserved;
|
||||
}
|
||||
|
||||
async get(userId: string, accountId: string): Promise<PortfolioDto> {
|
||||
const result = await this.read(userId, accountId);
|
||||
if (!result.data) {
|
||||
throw new ServiceUnavailableException({ code: TbankErrorCode.Unavailable });
|
||||
}
|
||||
return result.data;
|
||||
}
|
||||
|
||||
getAll(userId: string): Promise<PortfolioDto> {
|
||||
return this.get(userId, 'all');
|
||||
}
|
||||
|
||||
private async normalize(
|
||||
upstream: NonNullable<TbankPortfolio['positions']>,
|
||||
money: TbankMoney[],
|
||||
|
||||
@@ -44,15 +44,9 @@ export class ForecastRepository {
|
||||
data: { membershipRevision: { increment: 0 } },
|
||||
});
|
||||
if (!current.count) return null;
|
||||
}
|
||||
const connection = await tx.tbankConnection.findFirst({
|
||||
where: {
|
||||
...owner,
|
||||
membershipRevision: input.membershipRevision,
|
||||
invalidToken: false,
|
||||
},
|
||||
});
|
||||
if (!connection) return null;
|
||||
} else if (!(await tx.tbankConnection.findFirst({
|
||||
where: { ...owner, membershipRevision: input.membershipRevision, invalidToken: false },
|
||||
}))) return null;
|
||||
const key = { ...owner, scopeKey: input.scopeKey };
|
||||
const previous = await tx.tbankForecast.findUnique({
|
||||
where: { userId_revision_scopeKey: key },
|
||||
|
||||
@@ -52,7 +52,7 @@ export class PortfolioRepository {
|
||||
revision: string,
|
||||
accountId: string,
|
||||
processRunId: string,
|
||||
active: ReadonlySet<string>,
|
||||
pending: ReadonlyMap<string, Promise<void>>,
|
||||
force: boolean,
|
||||
) {
|
||||
return this.prisma.$transaction(async (tx) => {
|
||||
@@ -83,7 +83,7 @@ export class PortfolioRepository {
|
||||
if (
|
||||
previous?.state === 'running' &&
|
||||
previous.processRunId === processRunId &&
|
||||
active.has(previous.attemptId)
|
||||
pending.has(previous.attemptId)
|
||||
) {
|
||||
return { work: previous, launch: false };
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ import { Module } from '@nestjs/common';
|
||||
|
||||
import { TbankModule } from '../../../integrations/tbank/tbank.module';
|
||||
import { AuthModule } from '../../auth/auth.module';
|
||||
import { InstrumentSourcesModule } from '../instruments/instrument-sources.module';
|
||||
import { ForecastService } from './application/forecast.service';
|
||||
import { PortfolioService } from './application/portfolio.service';
|
||||
import { ForecastRepository } from './infrastructure/forecast.repository';
|
||||
@@ -10,7 +9,7 @@ import { PortfolioRepository } from './infrastructure/portfolio.repository';
|
||||
import { PortfolioController } from './presentation/portfolio.controller';
|
||||
|
||||
@Module({
|
||||
imports: [AuthModule, TbankModule, InstrumentSourcesModule],
|
||||
imports: [AuthModule, TbankModule],
|
||||
controllers: [PortfolioController],
|
||||
providers: [
|
||||
PortfolioService,
|
||||
|
||||
Reference in New Issue
Block a user