416 lines
13 KiB
TypeScript
416 lines
13 KiB
TypeScript
import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
|
import { randomUUID } from 'node:crypto';
|
|
|
|
import { StructuredLogger } from '../../../common/logging/structured-logger.service';
|
|
import { TbankErrorCode } from '../../../integrations/tbank/tbank-error-code';
|
|
import {
|
|
TbankAccessService,
|
|
TbankMembershipService,
|
|
type TbankSyncAttempt,
|
|
} from '../../../integrations/tbank/tbank.api';
|
|
import { PrismaService } from '../../../prisma/prisma.service';
|
|
import { BrokerageOperationsSyncService } from '../operations/operations.api';
|
|
import { PortfolioService } from '../portfolio/portfolio.api';
|
|
import { UpcomingEventsService } from '../upcoming-events/upcoming-events.service';
|
|
import {
|
|
HOURLY_CAUSE_INTERRUPTED,
|
|
type HourlyCause,
|
|
type HourlyChild,
|
|
type HourlySection,
|
|
TbankSyncRepository,
|
|
} from './tbank-sync.repository';
|
|
|
|
const HOUR_MS = 3_600_000;
|
|
const TICK_MS = 60_000;
|
|
const WORK_POLL_MS = 100;
|
|
const WORK_POLL_BACKOFF_FACTOR = 2;
|
|
const WORK_POLL_MAX_MS = 5_000;
|
|
// ponytail: fixed 15-minute step wait bound (900_000 ms), no config knob;
|
|
// make it configurable if slow steps become regular.
|
|
const WORK_WAIT_BOUND_MS = 900_000;
|
|
// An await that outlived WORK_WAIT_BOUND_MS. The underlying call keeps
|
|
// running; its late durable result is fenced like any other late publication.
|
|
const BOUND_EXPIRED = Symbol('hourly-bound-expired');
|
|
const FIRST_DAY_OF_MONTH = 1;
|
|
const NO_MILLISECONDS = 0;
|
|
|
|
type DispatchStep = (child: HourlyChild) => Promise<void>;
|
|
type Accounts = Awaited<ReturnType<TbankAccessService['local']>>['accounts'];
|
|
|
|
// One native timer per process. The timer only proposes work; the durable
|
|
// reservation in TbankSyncRepository is the fence that keeps a single
|
|
// attempt per connection. The deployment runs a single backend process.
|
|
// ponytail: no inter-process lease; add one before scaling past one process.
|
|
@Injectable()
|
|
export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
|
|
private readonly processRunId = randomUUID();
|
|
// Attempt ids of walks currently executing in this process; a running row
|
|
// outside this set lost its settle and is closed by recovery.
|
|
private readonly active = new Set<string>();
|
|
private timer: NodeJS.Timeout | null = null;
|
|
private ticking = false;
|
|
|
|
constructor(
|
|
private readonly prisma: PrismaService,
|
|
private readonly repository: TbankSyncRepository,
|
|
private readonly access: TbankAccessService,
|
|
private readonly membership: TbankMembershipService,
|
|
private readonly portfolio: PortfolioService,
|
|
private readonly operations: BrokerageOperationsSyncService,
|
|
private readonly events: UpcomingEventsService,
|
|
private readonly logger: StructuredLogger,
|
|
) {}
|
|
|
|
onModuleInit() {
|
|
this.timer = setInterval(() => void this.tick(), TICK_MS);
|
|
}
|
|
|
|
onModuleDestroy() {
|
|
if (this.timer) clearInterval(this.timer);
|
|
}
|
|
|
|
async tick() {
|
|
if (this.ticking) return;
|
|
this.ticking = true;
|
|
try {
|
|
const connections = await this.prisma.tbankConnection.findMany({
|
|
where: { invalidToken: false },
|
|
select: { userId: true },
|
|
});
|
|
for (const { userId } of connections) {
|
|
await this.launch(userId);
|
|
}
|
|
} catch {
|
|
// A failed tick only delays the proposal until the next tick.
|
|
} finally {
|
|
this.ticking = false;
|
|
}
|
|
}
|
|
|
|
// A fresh connection starts its first walk immediately, regardless of the
|
|
// previous term of the replaced connection.
|
|
private async launch(userId: string) {
|
|
await this.recoverAbandoned(userId);
|
|
const reserved = await this.repository.reserveHourly(
|
|
userId,
|
|
this.processRunId,
|
|
HOUR_MS,
|
|
);
|
|
if (!reserved) return null;
|
|
const { userId: owner, revision, id, attemptId } = reserved.work;
|
|
setImmediate(() => {
|
|
void this.walk({ userId: owner, revision, id, attemptId });
|
|
});
|
|
return reserved.work;
|
|
}
|
|
|
|
// A running hourly row of this process that has no active walk lost its
|
|
// settle (suppressed persistence error). Close it from its durable step
|
|
// results; rows of other processes are closed by the reservation restart
|
|
// branch. Rides the existing tick: no new timers or sanitors.
|
|
private async recoverAbandoned(userId: string) {
|
|
const rows = await this.repository.runningHourly(userId);
|
|
for (const row of rows) {
|
|
if (row.processRunId !== this.processRunId) continue;
|
|
if (this.active.has(row.attemptId)) continue;
|
|
await this.repository
|
|
.settleHourly(
|
|
{
|
|
userId,
|
|
revision: row.revision,
|
|
id: row.id,
|
|
attemptId: row.attemptId,
|
|
},
|
|
HOURLY_CAUSE_INTERRUPTED,
|
|
)
|
|
.catch(() => undefined);
|
|
}
|
|
}
|
|
|
|
// One membership pass, then the owners refresh their own datasets. The
|
|
// walk publishes only safe section results and holds no transaction or
|
|
// upstream call inside one. The attempt outcome is derived from durable
|
|
// step rows by the repository, never from walk-local state.
|
|
private async walk(attempt: TbankSyncAttempt) {
|
|
this.active.add(attempt.attemptId);
|
|
try {
|
|
const { userId, revision } = attempt;
|
|
const children: HourlyChild[] = [];
|
|
// Required sections whose dispatch failed never get a durable step
|
|
// row; they are recorded as problems so the attempt cannot succeed.
|
|
const dispatchProblems: HourlySection[] = [];
|
|
const dispatch: DispatchStep = async (child) => {
|
|
children.push(child);
|
|
await this.repository.recordHourlyChild(attempt, child);
|
|
};
|
|
const accounts = await this.refreshMembership(
|
|
attempt,
|
|
dispatch,
|
|
dispatchProblems,
|
|
);
|
|
if (!accounts) return;
|
|
await this.refreshAccounts(
|
|
userId,
|
|
revision,
|
|
accounts,
|
|
dispatch,
|
|
dispatchProblems,
|
|
);
|
|
await this.waitWorks(userId, children.map((child) => child.id));
|
|
for (const account of accounts) {
|
|
const pending = this.events
|
|
.get(userId, account.accountId)
|
|
.catch(() => undefined);
|
|
await this.withBound(pending);
|
|
}
|
|
await this.settle(attempt, null, dispatchProblems);
|
|
} catch {
|
|
await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED);
|
|
} finally {
|
|
this.active.delete(attempt.attemptId);
|
|
}
|
|
}
|
|
|
|
// Returns the current accounts, or null when the attempt is already
|
|
// settled (invalid token or a replaced connection revision).
|
|
private async refreshMembership(
|
|
attempt: TbankSyncAttempt,
|
|
dispatch: DispatchStep,
|
|
dispatchProblems: HourlySection[],
|
|
) {
|
|
const { userId, revision } = attempt;
|
|
const member = await this.withBound(
|
|
this.membership.request(userId, revision, true),
|
|
);
|
|
if (member && member !== BOUND_EXPIRED) {
|
|
await dispatch({ id: member.id, section: 'membership' });
|
|
const done = await this.withBound(member.done);
|
|
if (
|
|
done !== BOUND_EXPIRED &&
|
|
(await this.repository.workError(userId, member.id)) ===
|
|
TbankErrorCode.InvalidToken
|
|
) {
|
|
await this.settle(attempt);
|
|
return null;
|
|
}
|
|
} else {
|
|
// No durable step id (rejected request or an expired wait): record the
|
|
// observable problem; a late row stays fenced by its durable attempt id.
|
|
dispatchProblems.push({
|
|
section: 'membership',
|
|
errorCode: TbankErrorCode.Unavailable,
|
|
});
|
|
}
|
|
const current = await this.withBound(this.access.local(userId));
|
|
if (current === BOUND_EXPIRED || current.revision !== revision) {
|
|
await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED);
|
|
return null;
|
|
}
|
|
return current.accounts;
|
|
}
|
|
|
|
// Races a step await against the same per-step bound the row wait uses, so
|
|
// a dispatch that never resolves cannot hold the attempt running forever.
|
|
// The call itself is not cancelled; a rejection landing after the expiry is
|
|
// swallowed here because the expiry already accounts the step as a problem.
|
|
private withBound<T>(step: Promise<T>): Promise<T | typeof BOUND_EXPIRED> {
|
|
step.catch(() => undefined);
|
|
return new Promise<T | typeof BOUND_EXPIRED>((resolve, reject) => {
|
|
const timer = setTimeout(
|
|
() => resolve(BOUND_EXPIRED),
|
|
WORK_WAIT_BOUND_MS,
|
|
);
|
|
step.then(
|
|
(value) => {
|
|
clearTimeout(timer);
|
|
resolve(value);
|
|
},
|
|
(error: unknown) => {
|
|
clearTimeout(timer);
|
|
reject(error instanceof Error ? error : new Error(String(error)));
|
|
},
|
|
);
|
|
});
|
|
}
|
|
|
|
private async refreshAccounts(
|
|
userId: string,
|
|
revision: string,
|
|
accounts: Accounts,
|
|
dispatch: DispatchStep,
|
|
dispatchProblems: HourlySection[],
|
|
) {
|
|
for (const account of accounts) {
|
|
await this.dispatchPortfolio(
|
|
userId,
|
|
revision,
|
|
account.accountId,
|
|
dispatch,
|
|
dispatchProblems,
|
|
);
|
|
}
|
|
for (const account of accounts) {
|
|
await this.dispatchOperations(
|
|
userId,
|
|
revision,
|
|
account.accountId,
|
|
dispatch,
|
|
dispatchProblems,
|
|
);
|
|
}
|
|
for (const account of accounts) {
|
|
await this.dispatchEvents(
|
|
userId,
|
|
account.accountId,
|
|
dispatch,
|
|
dispatchProblems,
|
|
);
|
|
}
|
|
}
|
|
|
|
private async dispatchPortfolio(
|
|
userId: string,
|
|
revision: string,
|
|
accountId: string,
|
|
dispatch: DispatchStep,
|
|
dispatchProblems: HourlySection[],
|
|
) {
|
|
try {
|
|
const work = await this.withBound(
|
|
this.portfolio.request(userId, revision, accountId, true),
|
|
);
|
|
if (work && work !== BOUND_EXPIRED) {
|
|
await dispatch({ accountId, id: work.id, section: 'portfolio' });
|
|
return;
|
|
}
|
|
} catch {
|
|
// A failed dispatch has no durable step row; record the problem.
|
|
}
|
|
dispatchProblems.push({
|
|
accountId,
|
|
errorCode: TbankErrorCode.Unavailable,
|
|
section: 'portfolio',
|
|
});
|
|
}
|
|
|
|
private async dispatchOperations(
|
|
userId: string,
|
|
revision: string,
|
|
accountId: string,
|
|
dispatch: DispatchStep,
|
|
dispatchProblems: HourlySection[],
|
|
) {
|
|
try {
|
|
const id = await this.withBound(
|
|
this.operations.request(
|
|
userId,
|
|
[
|
|
{
|
|
accountId,
|
|
from: monthStartOf(new Date()),
|
|
to: new Date(),
|
|
},
|
|
],
|
|
true,
|
|
'screen',
|
|
revision,
|
|
),
|
|
);
|
|
if (id && id !== BOUND_EXPIRED) {
|
|
await dispatch({ accountId, id, section: 'operations' });
|
|
return;
|
|
}
|
|
} catch {
|
|
// A failed dispatch has no durable step row; record the problem.
|
|
}
|
|
dispatchProblems.push({
|
|
accountId,
|
|
errorCode: TbankErrorCode.Unavailable,
|
|
section: 'operations',
|
|
});
|
|
}
|
|
|
|
private async dispatchEvents(
|
|
userId: string,
|
|
accountId: string,
|
|
dispatch: DispatchStep,
|
|
dispatchProblems: HourlySection[],
|
|
) {
|
|
try {
|
|
const ids = await this.withBound(this.events.refresh(userId, accountId));
|
|
if (ids !== BOUND_EXPIRED) {
|
|
for (const id of ids) {
|
|
await dispatch({ accountId, id, section: 'events' });
|
|
}
|
|
return;
|
|
}
|
|
} catch {
|
|
// A failed dispatch has no durable step row; record the problem.
|
|
}
|
|
dispatchProblems.push({
|
|
accountId,
|
|
errorCode: TbankErrorCode.Unavailable,
|
|
section: 'events',
|
|
});
|
|
}
|
|
|
|
private async settle(
|
|
attempt: TbankSyncAttempt,
|
|
cause: HourlyCause | null = null,
|
|
dispatchProblems: HourlySection[] = [],
|
|
) {
|
|
// A suppressed settle error is not lost: the walk leaves the active set
|
|
// and the next tick closes the attempt from its durable step results.
|
|
await this.repository
|
|
.settleHourly(attempt, cause, dispatchProblems)
|
|
.catch(() => {
|
|
this.logger.event('error', 'tbank.hourly.settle_failed', {
|
|
userId: attempt.userId,
|
|
});
|
|
return undefined;
|
|
});
|
|
}
|
|
|
|
// Drops settled and missing rows from the wait set and expires waits that
|
|
// passed the per-step bound.
|
|
private advanceWaiting(
|
|
waiting: Map<string, number>,
|
|
rows: { id: string; state: string }[],
|
|
) {
|
|
for (const row of rows) {
|
|
if (row.state !== 'running') waiting.delete(row.id);
|
|
}
|
|
for (const id of [...waiting.keys()]) {
|
|
if (!rows.some((row) => row.id === id)) {
|
|
waiting.delete(id);
|
|
continue;
|
|
}
|
|
const startedAt = waiting.get(id) ?? NO_MILLISECONDS;
|
|
if (Date.now() - startedAt >= WORK_WAIT_BOUND_MS) waiting.delete(id);
|
|
}
|
|
}
|
|
|
|
// Waits for every dispatched step to leave the running state. Each wait is
|
|
// bounded so a stuck step cannot hold the attempt running forever; the
|
|
// bound does not cancel the upstream call, and late results stay in the
|
|
// step row where the attempt fence decides whether they count.
|
|
private async waitWorks(userId: string, ids: string[]) {
|
|
const waiting = new Map(ids.map((id) => [id, Date.now()]));
|
|
let pollMs = WORK_POLL_MS;
|
|
while (waiting.size) {
|
|
const rows = await this.repository.works(userId, [...waiting.keys()]);
|
|
this.advanceWaiting(waiting, rows);
|
|
if (waiting.size) {
|
|
await new Promise((resolve) => setTimeout(resolve, pollMs));
|
|
pollMs = Math.min(pollMs * WORK_POLL_BACKOFF_FACTOR, WORK_POLL_MAX_MS);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
function monthStartOf(now: Date) {
|
|
return new Date(
|
|
Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), FIRST_DAY_OF_MONTH),
|
|
);
|
|
}
|