6 Commits
Author SHA1 Message Date
ksv741 34f866fcec test: align resource availability metadata
Backend verify / commits (pull_request) Successful in 34s
Backend verify / security (pull_request) Successful in 13s
Backend verify / verify (pull_request) Successful in 5m38s
2026-10-09 22:21:22 +03:00
ksv741 20fb2f6c6e fix: keep candle retries read-only
Backend verify / commits (pull_request) Successful in 33s
Backend verify / security (pull_request) Successful in 1m28s
Backend verify / verify (pull_request) Failing after 4m12s
2026-10-09 21:36:27 +03:00
ksv741 92c76e5d4b test: align all accounts metadata
Backend verify / commits (pull_request) Successful in 36s
Backend verify / security (pull_request) Successful in 17s
Backend verify / verify (pull_request) Failing after 4m24s
2026-10-09 21:10:13 +03:00
ksv741 97ad968087 chore: prune obsolete eslint suppressions
Backend verify / commits (pull_request) Successful in 35s
Backend verify / security (pull_request) Successful in 22s
Backend verify / verify (pull_request) Failing after 3m34s
2026-10-09 20:57:56 +03:00
ksv741 3bc5f49866 test: align upcoming events with local reads
Backend verify / commits (pull_request) Successful in 37s
Backend verify / security (pull_request) Successful in 18s
Backend verify / verify (pull_request) Failing after 2m5s
2026-10-09 20:41:38 +03:00
ksv741 aee330c6c2 feat: reconcile hourly investment resource states 2026-10-09 19:56:14 +03:00
33 changed files with 126 additions and 2068 deletions
+12 -29
View File
@@ -7,19 +7,11 @@
checkout. Внешний API задаётся предоставленным публичным контрактом; checkout. Внешний API задаётся предоставленным публичным контрактом;
исходники или инструкции его реализации не являются зависимостью разработки. исходники или инструкции его реализации не являются зависимостью разработки.
При нехватке требования сообщи, какое решение не определено, до его реализации. При нехватке требования сообщи, какое решение не определено, до его реализации.
При конфликте с общим шаблоном следуй местным правилам и явному запросу
пользователя. Выбирай реализацию самостоятельно, но изменение финансовой
логики, публичного контракта или утверждённой семантики возвращай владельцу
на согласование до зависимой работы.
Начни с результата и affected areas. До edits проследи flow/invariants, найди Начни с результата и affected areas. До edits проследи flow/invariants, найди
owner и 1–3 примера. Код показывает механизм; случайный аналог не меняет контракт. owner и 1–3 примера. Код показывает механизм; случайный аналог не меняет контракт.
Выбери lightweight или SDD по Engineering Workflow из AGENTS; OpenSpec lifecycle Работай напрямую; OpenSpec/SDD lifecycle — только по явному запросу.
начинай после явного решения Change Owner. Перед Node-командами активируй местную `.nvmrc`: `nvm use --silent`.
Перед Node-командами активируй местную `.nvmrc`: если `nvm` не загружен в
non-interactive shell, загрузи установленный `$HOME/.nvm/nvm.sh`; затем
выполни `nvm use --silent` из backend. Если `nvm.sh` недоступен, остановись,
не используй системный Node.
Сохраняй чужие staged/unstaged/untracked изменения и read-only архивы. Сохраняй чужие staged/unstaged/untracked изменения и read-only архивы.
## Change scope ## Change scope
@@ -65,28 +57,19 @@ Peer-конфликт решай совместимыми версиями бе
## Проверки и завершение ## Проверки и завершение
| Когда | Что проверить | Code/dependency change: `npm ci`, полный `npm run verify` и дополнительные
| --- | --- | risk checks из местной testing convention. Targeted tests ускоряют итерацию,
| Окружение новое, изменились manifest/lockfile или Node, установка сомнительна | `npm ci` до Node-проверок; одна правка кода переустановки не требует. | но не заменяют финальный gate. `lint:fix`/format — edits, не proof.
| Итерация | Адресные тесты, typecheck и lint по затронутому поведению и риску. При изменении поведения добавь/обнови содержательную регрессию, если существующая проверка не ловит её; для косметики не создавай искусственный тест. | Markdown-only: проверь новые/изменённые тексты, локальные ссылки/якоря, маршруты
| Итоговый code/dependency diff | Перед объявлением готовности и передачей к интеграции выполни один полный `npm run verify` на итоговых входах. Дополнительные risk checks из testing convention и явно утверждённые task checks обязательны; CI проверяет независимо. | чтения и `git diff --check`; application build ради текста не нужен.
| Только Markdown | Проверь тексты, ссылки/якоря, маршруты чтения и `git diff --check`; application build не нужен. При изменении verification-контура проверь его работу. | Повтори routing по staged/unstaged/new files. Запиши краткие команды, результаты
и ограничения. Зелёный lint не заменяет semantic/security review.
Повтори routing по staged/unstaged/new files. В отчёте по задаче/PR запиши
проверенные входы, команду, exit code, результат и ограничения. Повторяй
проверку при изменении значимых входов (включая config, dependencies и
generated artifacts), сомнении в evidence или явном запросе пользователя.
`lint:fix`/format — edits, не proof; lint не заменяет semantic/security review.
## Git и релизы ## Git и релизы
Работа — в ветке `<type>/<issue-number>-<slug>` от актуального `develop`, Работа — в `codex/<task>` от актуального `develop`, затем PR в `develop`.
затем PR в `develop`; для cross-repo ссылки укажи owner/repo#number.
Имя ветки и связь PR с реальной Issue проверяются вручную: commitlint
проверяет синтаксис commit/PR title, но не существование и соответствие Issue.
Сохраняй чужие изменения; перед commit просмотри staged paths и diff. Сохраняй чужие изменения; перед commit просмотри staged paths и diff.
Commits и PR titles — Conventional Commits с parser-recognized Issue reference; допустим Commits и PR titles — Conventional Commits; допустим `spec:`. Каждый commit
`spec:`. Каждый commit содержит одну задачу. Исключения release и правила содержит одну связанную задачу. Root другого checkout не управляет этим repo.
ссылки — в Engineering Workflow из AGENTS. Root другого checkout не управляет этим repo.
Прямой push в `main`, force-push и перенос опубликованных тегов не входят в Прямой push в `main`, force-push и перенос опубликованных тегов не входят в
обычную работу. Перед release/hotfix прочитай местный README, раздел «Релизы». обычную работу. Перед release/hotfix прочитай местный README, раздел «Релизы».
-12
View File
@@ -1,12 +0,0 @@
---
name: Bug Report
about: Неверное поведение
---
## Проблема и шаги воспроизведения
## Факт / ожидание
## Acceptance Criteria
- [ ]
## Окружение / ссылки (если применимо)
-10
View File
@@ -1,10 +0,0 @@
---
name: Engineering Task
about: CI, тесты, документация, процесс и прочее
---
## Контекст и результат
## Scope / связанные Issues (если есть)
## Acceptance Criteria
- [ ]
-12
View File
@@ -1,12 +0,0 @@
---
name: Feature Request
about: Новое наблюдаемое поведение
---
## Зачем и какой результат
## Ожидаемое поведение
## Acceptance Criteria
- [ ]
## Зависимости / связанные Issues (если есть)
-12
View File
@@ -1,12 +0,0 @@
---
name: Refactoring / Technical Debt
about: Улучшение без нового поведения
---
## Проблема и результат
## Сохраняемые инварианты / границы
## Acceptance Criteria
- [ ]
## Зависимости / ссылки (если есть)
-11
View File
@@ -1,11 +0,0 @@
## Summary
## Linked Issue
Refs
## Changes
## Verification
- Команда / результат / ограничения:
## Breaking Changes (если есть)
+2 -6
View File
@@ -4,7 +4,7 @@ on:
push: push:
branches: [develop, main] branches: [develop, main]
pull_request: pull_request:
types: [opened, synchronize, reopened] types: [opened, synchronize, reopened, edited]
schedule: schedule:
- cron: '47 5 * * *' - cron: '47 5 * * *'
@@ -34,11 +34,7 @@ jobs:
HEAD_SHA: ${{ github.event.pull_request.head.sha }} HEAD_SHA: ${{ github.event.pull_request.head.sha }}
run: | run: |
printf '%s\n' "$PR_TITLE" | npm run commitlint -- --strict printf '%s\n' "$PR_TITLE" | npm run commitlint -- --strict
if [[ "$PR_TITLE" =~ ^chore\(release\):\ [0-9]+\.[0-9]+\.[0-9]+$ ]]; then npm run commitlint -- --strict --from "$BASE_SHA" --to "$HEAD_SHA"
npm run commitlint -- --strict --from "$HEAD_SHA^" --to "$HEAD_SHA"
else
npm run commitlint -- --strict --from "$BASE_SHA" --to "$HEAD_SHA"
fi
security: security:
needs: commits needs: commits
-3
View File
@@ -4,9 +4,6 @@
только его файлы и предоставленные публичные контракты. Перед работой прочитай только его файлы и предоставленные публичные контракты. Перед работой прочитай
[порядок работы](.ai/conventions/engineering.md#граница-и-порядок-работы), затем [порядок работы](.ai/conventions/engineering.md#граница-и-порядок-работы), затем
объединяй triggers задачи и путей; читай активированные темы до решения. объединяй triggers задачи и путей; читай активированные темы до решения.
До edits найди/создай подходящую Gitea Issue, проверь AC и ветку; чтение
допускается без Issue. Общий маршрут — [Engineering Workflow](https://git.personal.ksv741.ru/WealthLedger/main/src/commit/03b3b59e6ff2ba391f9dfc42a039a96d0d84c1c9/.ai/conventions/engineering-workflow.md)
(main#28; после merge — актуальная версия `develop`).
| Решение / сигнал | Читать | | Решение / сигнал | Читать |
| --- | --- | | --- | --- |
-38
View File
@@ -1,43 +1,5 @@
# Changelog # Changelog
## [2.1.0](https://git.personal.ksv741.ru/WealthLedger/backend/compare/v2.0.0...v2.1.0) (2026-10-12)
### Features
* add hourly portfolio sync history ([0057925](https://git.personal.ksv741.ru/WealthLedger/backend/commit/0057925888dde874830149865fc5e13920ac5783))
* add stock screener ([7d10ef8](https://git.personal.ksv741.ru/WealthLedger/backend/commit/7d10ef8adfce76ad2ed6368e1022fbd3f6847a23))
* reconcile hourly investment resource states ([323b83b](https://git.personal.ksv741.ru/WealthLedger/backend/commit/323b83b5827d1191fbc14cfdf1d0e62f249fc78d))
### Bug Fixes
* align backend workflow rules (WealthLedger/main[#27](https://git.personal.ksv741.ru/WealthLedger/backend/issues/27)) ([c7d0600](https://git.personal.ksv741.ru/WealthLedger/backend/commit/c7d060023b612af9e1dd4c2a2fb2d5d1bf173784))
* bound hourly step awaits by the per-step wait limit ([#56](https://git.personal.ksv741.ru/WealthLedger/backend/issues/56)) ([52fa3a3](https://git.personal.ksv741.ru/WealthLedger/backend/commit/52fa3a36baa5ab236a7ebd64ef1b4f447c91e129))
* keep candle retries read-only ([947ffef](https://git.personal.ksv741.ru/WealthLedger/backend/commit/947ffefa9b8cad7f19355c187f2bd50c9739a822))
* restore terminal hourly lifecycle after persistence failure ([#56](https://git.personal.ksv741.ru/WealthLedger/backend/issues/56)) ([fa57a20](https://git.personal.ksv741.ru/WealthLedger/backend/commit/fa57a201fdd8127774aee9af553da75a75add07c))
* unblock backend dependency audit ([85858a1](https://git.personal.ksv741.ru/WealthLedger/backend/commit/85858a160f00dd428c80b8fa1824c1779245fdfe))
### Documentation
* clarify backend verification policy ([d7895f2](https://git.personal.ksv741.ru/WealthLedger/backend/commit/d7895f287ee802e7b13df7a2a4c6dba0b711e5a9))
* pin backend workflow reference (WealthLedger/main[#27](https://git.personal.ksv741.ru/WealthLedger/backend/issues/27)) ([c5248db](https://git.personal.ksv741.ru/WealthLedger/backend/commit/c5248dbefc4b11a9f417bc835041a4294bd61047))
### CI
* skip checks on PR metadata edits ([0ad84ff](https://git.personal.ksv741.ru/WealthLedger/backend/commit/0ad84ff4f0b4712abc7e69dc299856f14af6eadd))
### Tests
* align all accounts metadata ([db375cb](https://git.personal.ksv741.ru/WealthLedger/backend/commit/db375cbee7d8f1eff48822b8e3c74898f40e8886))
* align resource availability metadata ([bc631e7](https://git.personal.ksv741.ru/WealthLedger/backend/commit/bc631e72de0d9b0ff7b261b1ea644a73c2943d2a))
* align upcoming events with local reads ([966b2a5](https://git.personal.ksv741.ru/WealthLedger/backend/commit/966b2a5836cdaf5b761e3c76568ff30a90ba1f4c))
* extend ledger sqlite timeout ([4870d49](https://git.personal.ksv741.ru/WealthLedger/backend/commit/4870d4950303ad1de743e19b665b2fbdc05b781a))
* increase ledger request timeout ([2facd86](https://git.personal.ksv741.ru/WealthLedger/backend/commit/2facd8675c74a97d92e0e8f44ed4a93aabb47674))
### Maintenance
* align backend engineering workflow (WealthLedger/main[#27](https://git.personal.ksv741.ru/WealthLedger/backend/issues/27)) ([a31e8d2](https://git.personal.ksv741.ru/WealthLedger/backend/commit/a31e8d2c4d85a21cdd5efdcc9f5fe503f210804b))
* prune obsolete eslint suppressions ([a46a411](https://git.personal.ksv741.ru/WealthLedger/backend/commit/a46a4112cbcc72b1c3ff17cac4ac8bebcf579bca))
## [2.0.0](https://git.personal.ksv741.ru/WealthLedger/backend/compare/v1.0.0...v2.0.0) (2026-10-07) ## [2.0.0](https://git.personal.ksv741.ru/WealthLedger/backend/compare/v1.0.0...v2.0.0) (2026-10-07)
### ⚠ BREAKING CHANGES ### ⚠ BREAKING CHANGES
+2 -4
View File
@@ -295,11 +295,9 @@ Durability оставляет временные DB, логи и `result.json`
## Релизы ## Релизы
Репозиторий выпускается независимо. `develop` — интеграция, `main` — принятые Репозиторий выпускается независимо. `develop` — интеграция, `main` — принятые
релизы. Обычная работа: issue-linked ветка от актуального `develop`, PR в `develop`; релизы. Работа: `codex/<task>` от актуального `develop`, PR в `develop`.
общие правила — в Engineering Workflow по ссылке из AGENTS.
Обычные PR допускают squash; release и обратная синхронизация сохраняют историю. Обычные PR допускают squash; release и обратная синхронизация сохраняют историю.
Commits и PR titles — Conventional Commits со ссылкой на Issue (кроме release), Commits и PR titles — Conventional Commits, включая `spec:`. Версии вручную не повышай.
включая `spec:`. Версии вручную не повышай.
1. Получи `main`, `develop` и теги: `git fetch origin --tags`. Убедись, что 1. Получи `main`, `develop` и теги: `git fetch origin --tags`. Убедись, что
`develop` содержит последний релиз. Создай `release/<name>` от проверенного `develop` содержит последний релиз. Создай `release/<name>` от проверенного
+1 -9
View File
@@ -1,12 +1,4 @@
module.exports = { module.exports = {
extends: ['@commitlint/config-conventional'], extends: ['@commitlint/config-conventional'],
defaultIgnores: false, rules: { 'type-enum': [2, 'always', ['feat', 'fix', 'perf', 'docs', 'spec', 'refactor', 'build', 'ci', 'test', 'style', 'chore', 'revert']] },
ignores: [
(message) => /^chore\(release\): \d+\.\d+\.\d+$/.test(message.trim()),
(message) => /^Merge pull request ['"].+['"] \(#\d+\) from .+ into .+$/.test(message.trim()),
],
rules: {
'type-enum': [2, 'always', ['feat', 'fix', 'perf', 'docs', 'spec', 'refactor', 'build', 'ci', 'test', 'style', 'chore', 'revert']],
'references-empty': [2, 'never'],
},
}; };
+2 -2
View File
@@ -1,12 +1,12 @@
{ {
"name": "wealth-ledger-backend", "name": "wealth-ledger-backend",
"version": "2.1.0", "version": "2.0.0",
"lockfileVersion": 3, "lockfileVersion": 3,
"requires": true, "requires": true,
"packages": { "packages": {
"": { "": {
"name": "wealth-ledger-backend", "name": "wealth-ledger-backend",
"version": "2.1.0", "version": "2.0.0",
"license": "UNLICENSED", "license": "UNLICENSED",
"dependencies": { "dependencies": {
"@grpc/grpc-js": "1.14.5", "@grpc/grpc-js": "1.14.5",
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "wealth-ledger-backend", "name": "wealth-ledger-backend",
"version": "2.1.0", "version": "2.0.0",
"description": "", "description": "",
"author": "", "author": "",
"private": true, "private": true,
@@ -14,7 +14,6 @@ describe('StructuredLogger', () => {
const logger = new StructuredLogger(pino as never); const logger = new StructuredLogger(pino as never);
logger.info('event', { logger.info('event', {
asOf: '2026-10-10T12:00:00.000Z',
userId: 'user', userId: 'user',
brokerageAccountId: 'account', brokerageAccountId: 'account',
password: 'secret', password: 'secret',
@@ -22,7 +21,6 @@ describe('StructuredLogger', () => {
expect(pino.info).toHaveBeenCalledWith( expect(pino.info).toHaveBeenCalledWith(
{ {
asOf: '2026-10-10T12:00:00.000Z',
userId: 'user', userId: 'user',
brokerageAccountId: 'account', brokerageAccountId: 'account',
}, },
@@ -5,7 +5,6 @@ type LogLevel = 'debug' | 'error' | 'info' | 'warn';
type LogContext = Record<string, boolean | number | string | undefined>; type LogContext = Record<string, boolean | number | string | undefined>;
const allowedFields = new Set([ const allowedFields = new Set([
'asOf',
'attempts', 'attempts',
'brokerageAccountId', 'brokerageAccountId',
'clientRequestId', 'clientRequestId',
@@ -133,18 +133,6 @@ export class TbankAccessService {
return this.tokens.decrypt(connection.encryptedToken); return this.tokens.decrypt(connection.encryptedToken);
} }
async current(userId: string): Promise<{ revision: string; token: string }> {
const connection = await this.prisma.tbankConnection.findUnique({
where: { userId },
select: { encryptedToken: true, invalidToken: true, revision: true },
});
if (!connection) throw new ConflictException({ code: TbankErrorCode.NotConnected });
if (connection.invalidToken) {
throw new BadRequestException({ code: TbankErrorCode.InvalidToken });
}
return { revision: connection.revision, token: this.tokens.decrypt(connection.encryptedToken) };
}
async publishMembership( async publishMembership(
attempt: TbankSyncAttempt, attempt: TbankSyncAttempt,
accounts: TbankAccount[], accounts: TbankAccount[],
@@ -8,9 +8,7 @@ import type {
export type TbankShare = TbankInstrument & { export type TbankShare = TbankInstrument & {
apiTradeAvailableFlag?: boolean; apiTradeAvailableFlag?: boolean;
assetUid?: string; assetUid?: string;
buyAvailableFlag?: boolean;
classCode?: string; classCode?: string;
countryOfRisk?: string;
currency?: string; currency?: string;
figi?: string; figi?: string;
first_1dayCandleDate?: TbankTimestamp; first_1dayCandleDate?: TbankTimestamp;
-4
View File
@@ -5,10 +5,6 @@ export enum InstrumentIdType {
Uid = 'INSTRUMENT_ID_TYPE_UID', Uid = 'INSTRUMENT_ID_TYPE_UID',
} }
export enum TbankInstrumentStatus {
All = 'INSTRUMENT_STATUS_ALL',
}
export enum TbankInstrumentType { export enum TbankInstrumentType {
Bond = 'INSTRUMENT_TYPE_BOND', Bond = 'INSTRUMENT_TYPE_BOND',
Share = 'INSTRUMENT_TYPE_SHARE', Share = 'INSTRUMENT_TYPE_SHARE',
-138
View File
@@ -282,141 +282,3 @@ describe('TbankClient metadata retries', () => {
expect(jest.getTimerCount()).toBe(0); expect(jest.getTimerCount()).toBe(0);
}); });
}); });
describe('TbankClient share catalog', () => {
const token = 'token';
const catalogSize = 40;
let client: TbankClient;
let call: jest.Mock;
let logger: {
event: jest.Mock<void, [string, string, Record<string, number | string>]>;
};
beforeEach(() => {
call = jest.fn();
logger = {
event: jest.fn<void, [string, string, Record<string, number | string>]>(),
};
client = Object.assign(
Object.create(TbankClient.prototype) as TbankClient,
{ call, logger },
);
});
it('fetches the full share catalog through Shares with INSTRUMENT_STATUS_ALL', async () => {
const instruments = [{ ticker: 'AFLT', uid: 'uid-aflt' }];
call.mockResolvedValue({ instruments });
await expect(client.getAllShares(token)).resolves.toEqual(instruments);
expect(call).toHaveBeenCalledWith('token', undefined, 'shares', {
instrumentStatus: 'INSTRUMENT_STATUS_ALL',
});
});
it('returns the whole catalog including a fitting instrument beyond the first 30 text matches', async () => {
const catalog = Array.from({ length: catalogSize }, (_, index) => {
const fitting = index === catalogSize - 1;
return {
apiTradeAvailableFlag: fitting,
buyAvailableFlag: fitting,
ticker: fitting ? 'FIT' : `T${index}`,
uid: fitting ? 'uid-fit' : `uid-${index}`,
};
});
call.mockResolvedValue({ instruments: catalog });
const shares = await client.getAllShares(token);
expect(shares).toEqual(catalog);
expect(shares).toHaveLength(catalogSize);
expect(shares).toContainEqual({
apiTradeAvailableFlag: true,
buyAvailableFlag: true,
ticker: 'FIT',
uid: 'uid-fit',
});
});
it('rejects when the source refuses instead of returning an empty success', async () => {
const error = new ServiceUnavailableException({
code: 'TBANK_UNAVAILABLE',
});
call.mockRejectedValue(error);
await expect(client.getAllShares(token)).rejects.toBe(error);
});
it('rejects when the response carries no instrument list', async () => {
call.mockResolvedValue({});
await expect(client.getAllShares(token)).rejects.toBeInstanceOf(
ServiceUnavailableException,
);
});
it('rejects an empty instrument list with the public unavailable code', async () => {
call.mockResolvedValue({ instruments: [] });
const failure = await client
.getAllShares(token)
.catch((error: unknown) => error);
expect(failure).toBeInstanceOf(ServiceUnavailableException);
expect((failure as ServiceUnavailableException).getResponse()).toEqual({
code: 'TBANK_UNAVAILABLE',
});
});
it('rejects a non-list instrument payload as uninterpretable', async () => {
call.mockResolvedValue({ instruments: 'corrupted' });
await expect(client.getAllShares(token)).rejects.toBeInstanceOf(
ServiceUnavailableException,
);
});
it('rejects an uninterpretable entry without instrument identity', async () => {
call.mockResolvedValue({
instruments: [
{ ticker: 'FIT', uid: 'uid-fit' },
{ ticker: 'BROKEN' },
],
});
await expect(client.getAllShares(token)).rejects.toBeInstanceOf(
ServiceUnavailableException,
);
});
it('registers asOf and source instrument count on a successful catalog fetch', async () => {
call.mockResolvedValue({
instruments: [{ uid: 'uid-a' }, { uid: 'uid-b' }],
});
await client.getAllShares(token);
expect(logger.event).toHaveBeenCalledWith(
'info',
'tbank.shares.catalog.fetched',
expect.objectContaining({ count: 2, operation: 'shares' }),
);
expect(logger.event.mock.calls[0][2].asOf).toEqual(expect.any(String));
});
it('logs a distinguishable failure without secrets when the catalog is unusable', async () => {
call.mockResolvedValue({ instruments: [] });
await expect(client.getAllShares('secret-token')).rejects.toBeInstanceOf(
ServiceUnavailableException,
);
expect(logger.event).toHaveBeenCalledWith(
'error',
'tbank.shares.catalog.unusable',
expect.objectContaining({ operation: 'shares' }),
);
expect(JSON.stringify(logger.event.mock.calls)).not.toContain(
'secret-token',
);
});
});
-30
View File
@@ -34,7 +34,6 @@ import {
InstrumentIdType, InstrumentIdType,
TbankAccountStatus, TbankAccountStatus,
TbankCandleInterval, TbankCandleInterval,
TbankInstrumentStatus,
TbankInstrumentType, TbankInstrumentType,
TbankOperationState, TbankOperationState,
} from './tbank-protocol'; } from './tbank-protocol';
@@ -326,35 +325,6 @@ export class TbankClient {
).then((response) => response.instruments ?? []); ).then((response) => response.instruments ?? []);
} }
async getAllShares(token: string): Promise<TbankShare[]> {
const response = await this.call<{ instruments?: TbankShare[] }>(
token,
this.instruments,
'shares',
{ instrumentStatus: TbankInstrumentStatus.All },
);
const { instruments } = response;
if (
!Array.isArray(instruments) ||
!instruments.length ||
instruments.some((item) => !item?.uid)
) {
this.logger.event('error', 'tbank.shares.catalog.unusable', {
code: TbankErrorCode.Unavailable,
operation: 'shares',
});
throw new ServiceUnavailableException({
code: TbankErrorCode.Unavailable,
});
}
this.logger.event('info', 'tbank.shares.catalog.fetched', {
asOf: new Date().toISOString(),
count: instruments.length,
operation: 'shares',
});
return instruments;
}
findBonds(token: string, query: string): Promise<TbankBond[]> { findBonds(token: string, query: string): Promise<TbankBond[]> {
return this.call<{ instruments?: TbankBond[] }>( return this.call<{ instruments?: TbankBond[] }>(
token, token,
@@ -58,7 +58,7 @@ export class CatalogEnrichmentService {
private running = false; private running = false;
private readonly completions = new Map< private readonly completions = new Map<
string, string,
{ attempt: TbankSyncAttempt; done: Promise<void>; resolve: () => void; workId: string } { done: Promise<void>; resolve: () => void; workId: string }
>(); >();
async ensureDetail( async ensureDetail(
@@ -208,7 +208,7 @@ export class CatalogEnrichmentService {
const done = new Promise<void>((finish) => { const done = new Promise<void>((finish) => {
resolve = finish; resolve = finish;
}); });
this.completions.set(attemptId, { attempt, done, resolve, workId: id }); this.completions.set(attemptId, { workId: id, done, resolve });
const key = producerKey(input); const key = producerKey(input);
const producer = this.producers.get(key); const producer = this.producers.get(key);
if (producer) producer.observers.push({ attempt, input }); if (producer) producer.observers.push({ attempt, input });
@@ -580,14 +580,6 @@ export class CatalogEnrichmentService {
} }
} }
} catch { } catch {
// Persistence failed mid-batch: finish every still-open observer
// attempt durably so no source row is left running. Best effort — a
// row that stays open is re-reserved by the next reader.
for (const completion of this.completions.values()) {
await this.repository
.finish(completion.attempt, false)
.catch(() => undefined);
}
for (const completion of this.completions.values()) completion.resolve(); for (const completion of this.completions.values()) completion.resolve();
this.completions.clear(); this.completions.clear();
this.active.clear(); this.active.clear();
@@ -1,6 +1,5 @@
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { TbankAccessService } from '../../../integrations/tbank/tbank-access.service';
import { TbankModule } from '../../../integrations/tbank/tbank.module'; import { TbankModule } from '../../../integrations/tbank/tbank.module';
import { AuthModule } from '../../auth/auth.module'; import { AuthModule } from '../../auth/auth.module';
import { PortfolioModule } from '../portfolio/portfolio.module'; import { PortfolioModule } from '../portfolio/portfolio.module';
@@ -13,7 +12,6 @@ import { GetStockCompanyUseCase } from './stocks/application/get-stock-company.u
import { StockCatalogService } from './stocks/application/stock-catalog.service'; import { StockCatalogService } from './stocks/application/stock-catalog.service';
import { StockIncomeService } from './stocks/application/stock-income.service'; import { StockIncomeService } from './stocks/application/stock-income.service';
import { StockMarketService } from './stocks/application/stock-market.service'; import { StockMarketService } from './stocks/application/stock-market.service';
import { StockScreenerService } from './stocks/application/stock-screener.service';
import { StockController } from './stocks/presentation/stock.controller'; import { StockController } from './stocks/presentation/stock.controller';
@Module({ @Module({
@@ -21,14 +19,12 @@ import { StockController } from './stocks/presentation/stock.controller';
controllers: [StockController, BondController], controllers: [StockController, BondController],
exports: [StockIncomeService, BondIncomeService], exports: [StockIncomeService, BondIncomeService],
providers: [ providers: [
{ provide: 'TBANK_ACCESS', useExisting: TbankAccessService },
BondIncomeService, BondIncomeService,
BondMarketService, BondMarketService,
GetStockCompanyUseCase, GetStockCompanyUseCase,
StockIncomeService, StockIncomeService,
StockMarketService, StockMarketService,
StockCatalogService, StockCatalogService,
StockScreenerService,
BondCatalogService, BondCatalogService,
], ],
}) })
@@ -1,106 +0,0 @@
import { NotFoundException, ServiceUnavailableException } from '@nestjs/common';
import type { TbankShare } from '../../../../../integrations/tbank/tbank.api';
import { StockScreenerService } from './stock-screener.service';
const token = 'secret-token';
const revision = 'revision-1';
const pageSize = 30;
function share(index: number, overrides: Partial<TbankShare> = {}): TbankShare {
return {
apiTradeAvailableFlag: true,
buyAvailableFlag: true,
currency: 'RUB',
lot: 1,
name: `Share ${index}`,
ticker: `T${String(index).padStart(2, '0')}`,
uid: `e6123145-9665-43e0-8413-cd61b8aa${String(index).padStart(4, '0')}`,
...overrides,
};
}
describe('StockScreenerService', () => {
const access = { current: jest.fn(), local: jest.fn() };
const client = { getAllShares: jest.fn() };
const logger = { event: jest.fn() };
let service: StockScreenerService;
beforeEach(() => {
jest.clearAllMocks();
access.current.mockResolvedValue({ revision, token });
access.local.mockResolvedValue({ revision });
client.getAllShares.mockResolvedValue(
Array.from({ length: pageSize + 1 }, (_, index) => share(index)),
);
service = new StockScreenerService(access, client as never, logger as never);
});
it('stores the full selected catalog and returns stable non-overlapping pages', async () => {
const first = await service.search('user-1', {});
const second = await service.page('user-1', first.searchId, 2);
expect(first.items).toHaveLength(pageSize);
expect(first.hasNext).toBe(true);
expect(second).toMatchObject({ asOf: first.asOf, hasNext: false, page: 2, searchId: first.searchId });
expect(second.items).toEqual([expect.objectContaining({ ticker: 'T30' })]);
expect(client.getAllShares).toHaveBeenCalledTimes(1);
});
it('creates a fresh result for every search and preserves a successful empty result', async () => {
client.getAllShares.mockResolvedValueOnce([share(1, { sector: 'energy' })]).mockResolvedValueOnce([share(2, { sector: 'energy' })]);
const first = await service.search('user-1', { sector: 'finance' });
const second = await service.search('user-1', { sector: 'finance' });
expect(first).toMatchObject({ hasNext: false, items: [], page: 1 });
expect(second).toMatchObject({ hasNext: false, items: [], page: 1 });
expect(second.searchId).not.toBe(first.searchId);
expect(client.getAllShares).toHaveBeenCalledTimes(2);
});
it('does not store a result when the catalog source fails', async () => {
client.getAllShares.mockRejectedValueOnce(new ServiceUnavailableException());
await expect(service.search('user-1', {})).rejects.toBeInstanceOf(ServiceUnavailableException);
});
it('denies missing, foreign, stale, and out-of-range pages without fetching', async () => {
const first = await service.search('user-1', {});
access.local.mockResolvedValueOnce({ revision: 'revision-2' });
await expect(service.page('user-1', first.searchId, 1)).rejects.toBeInstanceOf(NotFoundException);
await expect(service.page('user-2', first.searchId, 1)).rejects.toBeInstanceOf(NotFoundException);
await expect(service.page('user-1', 'missing', 1)).rejects.toBeInstanceOf(NotFoundException);
await expect(service.page('user-1', first.searchId, 3)).rejects.toMatchObject({ status: 400 });
expect(client.getAllShares).toHaveBeenCalledTimes(1);
});
it('null-normalizes optional source fields and rejects malformed required fields', async () => {
client.getAllShares.mockResolvedValueOnce([
share(1, {
countryOfRisk: undefined,
currency: undefined,
isin: undefined,
realExchange: undefined,
sector: undefined,
shareType: undefined,
ticker: undefined,
}),
]);
const result = await service.search('user-1', {});
expect(result.items[0]).toEqual(expect.objectContaining({ countryOfRisk: null, currency: null, isin: null, logoName: null, ticker: null }));
client.getAllShares.mockResolvedValueOnce([share(2, { name: ' ' })]);
await expect(service.search('user-1', {})).rejects.toBeInstanceOf(ServiceUnavailableException);
});
it('preserves the catalog logo name in screener results', async () => {
client.getAllShares.mockResolvedValueOnce([share(1, { brand: { logoName: 'sber.png' } })]);
const result = await service.search('user-1', {});
expect(result.items).toEqual([expect.objectContaining({ logoName: 'sber.png' })]);
});
});
@@ -1,140 +0,0 @@
import {
BadRequestException,
Inject,
Injectable,
NotFoundException,
ServiceUnavailableException,
} from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import type { TbankShare } from '../../../../../integrations/tbank/tbank-market.types';
import { StructuredLogger } from '../../../../../common/logging/structured-logger.service';
import { TbankClient } from '../../../../../integrations/tbank/tbank.client';
import {
selectStockScreenerInstruments,
type StockScreenerConditions,
} from '../domain/stock-screener-selection.rules';
const MAX_SNAPSHOTS = 100;
const PAGE_SIZE = 30;
const MIN_PAGE = 1;
const MINIMUM_SNAPSHOT_COUNT = 1;
const SECONDS_PER_MINUTE = 60;
const SNAPSHOT_TTL_MINUTES = 5;
const MILLISECONDS_PER_SECOND = 1_000;
const SNAPSHOT_TTL_MS = SNAPSHOT_TTL_MINUTES * SECONDS_PER_MINUTE * MILLISECONDS_PER_SECOND;
const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
type StockScreenerItem = {
countryOfRisk: null | string;
currency: null | string;
instrumentUid: string;
isin: null | string;
logoName: null | string;
lot: null | number;
name: string;
realExchange: null | string;
sector: null | string;
shareType: null | string;
ticker: null | string;
};
type StockScreenerPage = {
asOf: string;
hasNext: boolean;
items: StockScreenerItem[];
page: number;
searchId: string;
};
type Snapshot = { asOf: string; expiresAt: number; items: StockScreenerItem[]; revision: string; userId: string };
type TbankAccess = {
current(userId: string): Promise<{ revision: string; token: string }>;
local(userId: string): Promise<{ revision: string }>;
};
@Injectable()
export class StockScreenerService {
private readonly snapshots = new Map<string, Snapshot>();
constructor(
@Inject('TBANK_ACCESS') private readonly access: TbankAccess,
private readonly client: TbankClient,
private readonly logger: StructuredLogger,
) {}
async search(userId: string, conditions: StockScreenerConditions): Promise<StockScreenerPage> {
const connection = await this.access.current(userId);
const instruments = await this.client.getAllShares(connection.token);
const asOf = new Date().toISOString();
const selected = selectStockScreenerInstruments({ asOf: new Date(asOf), instruments }, conditions)
.instruments.map((instrument) => this.item(instrument));
const searchId = randomUUID();
this.evict();
this.snapshots.set(searchId, {
asOf,
expiresAt: Date.now() + SNAPSHOT_TTL_MS,
items: selected,
revision: connection.revision,
userId,
});
this.logger.event('info', 'stock.screener.search.completed', { asOf, count: selected.length, operation: 'stock-screener-search' });
return this.result(searchId, this.snapshots.get(searchId) as Snapshot, MIN_PAGE);
}
async page(userId: string, searchId: string, page: number): Promise<StockScreenerPage> {
const snapshot = this.snapshots.get(searchId);
if (!snapshot || snapshot.expiresAt <= Date.now()) {
this.snapshots.delete(searchId);
this.logger.event('info', 'stock.screener.result.missing', { operation: 'stock-screener-read' });
throw new NotFoundException();
}
if (snapshot.userId !== userId) {
this.logger.event('warn', 'stock.screener.result.rejected', { operation: 'stock-screener-read' });
throw new NotFoundException();
}
const current = await this.access.local(userId);
if (snapshot.revision !== current.revision) {
this.logger.event('warn', 'stock.screener.result.rejected', { operation: 'stock-screener-read' });
throw new NotFoundException();
}
return this.result(searchId, snapshot, page);
}
private result(searchId: string, snapshot: Snapshot, page: number): StockScreenerPage {
const start = (page - MIN_PAGE) * PAGE_SIZE;
if (!Number.isInteger(page) || page < MIN_PAGE || (start >= snapshot.items.length && !(page === MIN_PAGE && !snapshot.items.length))) {
throw new BadRequestException();
}
const items = snapshot.items.slice(start, start + PAGE_SIZE);
return { asOf: snapshot.asOf, hasNext: start + PAGE_SIZE < snapshot.items.length, items, page, searchId };
}
private item(instrument: TbankShare): StockScreenerItem {
if (!instrument.uid || !UUID_PATTERN.test(instrument.uid) || !instrument.name?.trim()) {
throw new ServiceUnavailableException();
}
return {
countryOfRisk: instrument.countryOfRisk ?? null,
currency: instrument.currency ?? null,
instrumentUid: instrument.uid,
isin: instrument.isin ?? null,
lot: instrument.lot ?? null,
logoName: instrument.brand?.logoName?.trim() || null,
name: instrument.name,
realExchange: instrument.realExchange ?? null,
sector: instrument.sector ?? null,
shareType: instrument.shareType ?? null,
ticker: instrument.ticker ?? null,
};
}
private evict(): void {
for (const [searchId, snapshot] of this.snapshots) {
if (snapshot.expiresAt <= Date.now()) this.snapshots.delete(searchId);
}
while (this.snapshots.size >= MAX_SNAPSHOTS) {
const searchId = this.snapshots.keys().next().value;
if (searchId && this.snapshots.size >= MINIMUM_SNAPSHOT_COUNT) this.snapshots.delete(searchId);
}
}
}
@@ -1,188 +0,0 @@
import type { TbankShare } from '../../../../../integrations/tbank/tbank.api';
import { selectStockScreenerInstruments } from './stock-screener-selection.rules';
const asOf = new Date('2026-10-10T12:00:00.000Z');
function share(overrides: Partial<TbankShare> = {}): TbankShare {
return {
apiTradeAvailableFlag: true,
buyAvailableFlag: true,
currency: 'RUB',
lot: 10,
name: 'North Star',
ticker: 'NSTR',
uid: 'uid-1',
...overrides,
};
}
describe('selectStockScreenerInstruments', () => {
it('keeps only instruments matching every selected condition', () => {
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [
share({ isin: 'A', sector: 'energy' }),
share({ currency: 'USD', isin: 'B', sector: 'energy', uid: 'uid-2' }),
share({ isin: 'C', name: 'Other', sector: 'energy', uid: 'uid-3' }),
share({ isin: 'D', sector: 'finance', uid: 'uid-4' }),
],
},
{ currency: 'RUB', query: 'star', sector: 'energy' },
);
expect(result).toEqual({ asOf, instruments: [share({ isin: 'A', sector: 'energy' })] });
});
it.each([
['sector', { sector: 'energy' }, share({ sector: undefined })],
['share type', { shareType: 'common' }, share({ shareType: 'preferred' })],
['currency', { currency: 'USD' }, share({ currency: undefined })],
['exchange', { realExchange: 'MOEX' }, share({ realExchange: undefined })],
['country of risk', { countryOfRisk: 'RU' }, share({ countryOfRisk: 'US' })],
['query', { query: 'star' }, share({ name: undefined, ticker: undefined })],
] as const)('excludes an unknown or unequal selected %s', (_condition, conditions, instrument) => {
const result = selectStockScreenerInstruments(
{ asOf, instruments: [instrument] },
conditions,
);
expect(result.instruments).toEqual([]);
});
it('uses an inclusive maximum lot and ignores lots when it is not selected', () => {
const instruments = [
share({ isin: 'equal', lot: 10, uid: 'equal' }),
share({ isin: 'large', lot: 11, uid: 'large' }),
share({ isin: 'unknown', lot: undefined, uid: 'unknown' }),
share({ isin: 'zero', lot: 0, uid: 'zero' }),
];
expect(
selectStockScreenerInstruments({ asOf, instruments }, { maxLot: 10 }).instruments.map(
({ uid }) => uid,
),
).toEqual(['equal']);
expect(
selectStockScreenerInstruments({ asOf, instruments }, {}).instruments.map(({ uid }) => uid),
).toEqual(['equal', 'large', 'unknown', 'zero']);
});
it('chooses the smallest matching uid as the complete representative of an ISIN', () => {
const chosen = share({ isin: ' same ', lot: 1, name: 'Chosen', uid: 'a' });
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [share({ isin: 'SAME', lot: 99, name: 'Sibling', uid: 'z' }), chosen],
},
{},
);
expect(result.instruments).toEqual([chosen]);
expect(result.instruments[0]).toMatchObject({ lot: 1, name: 'Chosen', uid: 'a' });
});
it('chooses the sole matching mode of an issue', () => {
const matching = share({ isin: 'same', sector: 'energy', uid: 'matching' });
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [share({ isin: 'same', sector: 'finance', uid: 'smaller-uid' }), matching],
},
{ sector: 'energy' },
);
expect(result.instruments).toEqual([matching]);
});
it('keeps different ISIN issues from one issuer separate', () => {
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [
share({ isin: 'common', shareType: 'common', uid: 'common' }),
share({ isin: 'preferred', shareType: 'preferred', uid: 'preferred' }),
],
},
{},
);
expect(result.instruments.map(({ uid }) => uid)).toEqual(['common', 'preferred']);
});
it('treats instruments without an ISIN as separate issues', () => {
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [share({ isin: ' ', uid: 'first' }), share({ isin: undefined, uid: 'second' })],
},
{},
);
expect(result.instruments.map(({ uid }) => uid)).toEqual(['first', 'second']);
});
it('preserves exact whitespace-differing UIDs as separate blank-ISIN issues', () => {
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [
share({ isin: ' ', uid: 'uid' }),
share({ isin: undefined, uid: ' uid ' }),
],
},
{},
);
expect(result.instruments.map(({ uid }) => uid)).toEqual([' uid ', 'uid']);
});
it('returns the exact chosen uid for the later card link', () => {
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [share({ isin: 'same', uid: 'sibling' }), share({ isin: 'SAME', uid: 'chosen' })],
},
{},
);
expect(result.instruments[0]?.uid).toBe('chosen');
});
it.each([
[true, true, 1],
[true, false, 0],
[false, true, 0],
[false, false, 0],
[true, undefined, 0],
[undefined, true, 0],
] as const)(
'includes only confirmed buy and API availability: buy=%s api=%s',
(buyAvailableFlag, apiTradeAvailableFlag, expectedCount) => {
const result = selectStockScreenerInstruments(
{ asOf, instruments: [share({ apiTradeAvailableFlag, buyAvailableFlag })] },
{},
);
expect(result.instruments).toHaveLength(expectedCount);
},
);
it('orders representatives by ticker and then uid before paging', () => {
const result = selectStockScreenerInstruments(
{
asOf,
instruments: [
share({ isin: 'c', ticker: 'B', uid: 'z' }),
share({ isin: 'a', ticker: 'A', uid: 'z' }),
share({ isin: 'b', ticker: 'A', uid: 'a' }),
share({ isin: 'empty', ticker: ' ', uid: 'empty' }),
],
},
{},
);
expect(result.instruments.map(({ uid }) => uid)).toEqual(['empty', 'a', 'z', 'z']);
});
});
@@ -1,106 +0,0 @@
import type { TbankShare } from '../../../../../integrations/tbank/tbank.api';
export type StockScreenerCatalog = {
asOf: Date;
instruments: TbankShare[];
};
export type StockScreenerConditions = {
countryOfRisk?: string;
currency?: string;
maxLot?: number;
query?: string;
realExchange?: string;
sector?: string;
shareType?: string;
};
type IdentifiedShare = TbankShare & { uid: string };
const sortBefore = 0;
export function selectStockScreenerInstruments(
catalog: StockScreenerCatalog,
conditions: StockScreenerConditions,
): StockScreenerCatalog {
const representatives = new Map<string, IdentifiedShare>();
for (const instrument of catalog.instruments) {
if (!matches(instrument, conditions)) continue;
const { uid } = instrument;
const isin = instrument.isin?.trim().toUpperCase();
const issueKey = isin ? `isin:${isin}` : `uid:${uid}`;
const existing = representatives.get(issueKey);
if (!existing || uid.localeCompare(existing.uid) < sortBefore) {
representatives.set(issueKey, instrument);
}
}
return {
asOf: catalog.asOf,
instruments: [...representatives.values()].sort(
(left, right) =>
(left.ticker?.trim() ?? '').localeCompare(right.ticker?.trim() ?? '') ||
left.uid.localeCompare(right.uid),
),
};
}
function matches(
instrument: TbankShare,
conditions: StockScreenerConditions,
): instrument is IdentifiedShare {
if (
!instrument.uid?.trim() ||
instrument.buyAvailableFlag !== true ||
instrument.apiTradeAvailableFlag !== true
) {
return false;
}
if (
conditions.query !== undefined &&
!`${instrument.name ?? ''}\n${instrument.ticker ?? ''}`
.toLowerCase()
.includes(conditions.query.toLowerCase())
) {
return false;
}
if (conditions.sector !== undefined && instrument.sector !== conditions.sector) {
return false;
}
if (
conditions.shareType !== undefined &&
instrument.shareType !== conditions.shareType
) {
return false;
}
if (
conditions.currency !== undefined &&
instrument.currency !== conditions.currency
) {
return false;
}
if (
conditions.realExchange !== undefined &&
instrument.realExchange !== conditions.realExchange
) {
return false;
}
if (
conditions.countryOfRisk !== undefined &&
instrument.countryOfRisk !== conditions.countryOfRisk
) {
return false;
}
const minimumPositiveLot = 0;
return (
conditions.maxLot === undefined ||
(Number.isInteger(instrument.lot) &&
instrument.lot !== undefined &&
instrument.lot > minimumPositiveLot &&
instrument.lot <= conditions.maxLot)
);
}
@@ -1,21 +1,15 @@
import type { Request } from 'express'; import type { Request } from 'express';
import { import {
BadRequestException,
Body,
Controller, Controller,
Get, Get,
HttpCode,
HttpStatus,
Param, Param,
ParseUUIDPipe, ParseUUIDPipe,
Post,
Query, Query,
Req, Req,
UseGuards, UseGuards,
ValidationPipe,
} from '@nestjs/common'; } from '@nestjs/common';
import { ApiBearerAuth, ApiOkResponse, ApiOperation, ApiTags } from '@nestjs/swagger'; import { ApiBearerAuth, ApiOkResponse, ApiTags } from '@nestjs/swagger';
import { ApiProblemResponses } from '../../../../../common/decorators/api-safe-error-response.decorator'; import { ApiProblemResponses } from '../../../../../common/decorators/api-safe-error-response.decorator';
import { AccessPayload, JwtAuthGuard } from '../../../../auth/auth.api'; import { AccessPayload, JwtAuthGuard } from '../../../../auth/auth.api';
@@ -24,7 +18,6 @@ import { GetStockCompanyUseCase } from '../application/get-stock-company.use-cas
import { StockCatalogService } from '../application/stock-catalog.service'; import { StockCatalogService } from '../application/stock-catalog.service';
import { StockIncomeService } from '../application/stock-income.service'; import { StockIncomeService } from '../application/stock-income.service';
import { StockMarketService } from '../application/stock-market.service'; import { StockMarketService } from '../application/stock-market.service';
import { StockScreenerService } from '../application/stock-screener.service';
import { import {
StockCandlesQueryDto, StockCandlesQueryDto,
StockCandlesResourceDto, StockCandlesResourceDto,
@@ -33,9 +26,6 @@ import {
StockPositionResourceDto, StockPositionResourceDto,
StockQuoteResourceDto, StockQuoteResourceDto,
StockResourceDto, StockResourceDto,
StockScreenerCriteriaDto,
StockScreenerPageDto,
StockScreenerPageQueryDto,
StockSearchQueryDto, StockSearchQueryDto,
StockSearchResourceDto, StockSearchResourceDto,
StockTickerParamsDto, StockTickerParamsDto,
@@ -52,7 +42,6 @@ export class StockController {
private readonly market: StockMarketService, private readonly market: StockMarketService,
private readonly income: StockIncomeService, private readonly income: StockIncomeService,
private readonly companyInfo: GetStockCompanyUseCase, private readonly companyInfo: GetStockCompanyUseCase,
private readonly screener: StockScreenerService,
) {} ) {}
@Get() @Get()
@@ -82,33 +71,6 @@ export class StockController {
); );
} }
@Post('screener/search')
@HttpCode(HttpStatus.OK)
@ApiOperation({ operationId: 'createStockScreenerSearch' })
@ApiOkResponse({ type: StockScreenerPageDto })
searchScreener(
@Req() request: Request & { user: AccessPayload },
@Body(new ValidationPipe({ forbidNonWhitelisted: true, forbidUnknownValues: true, whitelist: true }))
criteria: StockScreenerCriteriaDto,
) {
const allowed = new Set(['countryOfRisk', 'currency', 'maxLot', 'query', 'realExchange', 'sector', 'shareType']);
if (Object.keys(request.body as object).some((key) => !allowed.has(key))) {
throw new BadRequestException();
}
return this.screener.search(request.user.sub, criteria);
}
@Get('screener/search/:searchId')
@ApiOperation({ operationId: 'readStockScreenerPage' })
@ApiOkResponse({ type: StockScreenerPageDto })
pageScreener(
@Req() request: Request & { user: AccessPayload },
@Param('searchId') searchId: string,
@Query() query: StockScreenerPageQueryDto,
) {
return this.screener.page(request.user.sub, searchId, query.page);
}
@Get(':uid') @Get(':uid')
@ApiOkResponse({ type: StockResourceDto }) @ApiOkResponse({ type: StockResourceDto })
get( get(
@@ -2,12 +2,10 @@ import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
import { Transform } from 'class-transformer'; import { Transform } from 'class-transformer';
import { import {
IsIn, IsIn,
IsInt,
IsOptional, IsOptional,
IsString, IsString,
Matches, Matches,
MaxLength, MaxLength,
Min,
MinLength, MinLength,
} from 'class-validator'; } from 'class-validator';
@@ -23,86 +21,6 @@ import {
import { PortfolioPositionDto } from '../../../portfolio/portfolio.api'; import { PortfolioPositionDto } from '../../../portfolio/portfolio.api';
const MIN_TICKER_LENGTH = 1; const MIN_TICKER_LENGTH = 1;
const MAX_SCREENER_QUERY_LENGTH = 80;
const MIN_PAGE = 1;
export class StockScreenerCriteriaDto {
@ApiPropertyOptional({ maxLength: MAX_SCREENER_QUERY_LENGTH })
@Transform(({ value }: { value: unknown }) =>
value === '' ? undefined : value,
)
@IsOptional()
@IsString()
@MinLength(MIN_TICKER_LENGTH)
@MaxLength(MAX_SCREENER_QUERY_LENGTH)
query?: string;
@ApiPropertyOptional({ minLength: MIN_TICKER_LENGTH })
@IsOptional()
@IsString()
@MinLength(MIN_TICKER_LENGTH)
sector?: string;
@ApiPropertyOptional({ minLength: MIN_TICKER_LENGTH })
@IsOptional()
@IsString()
@MinLength(MIN_TICKER_LENGTH)
shareType?: string;
@ApiPropertyOptional({ minLength: MIN_TICKER_LENGTH })
@IsOptional()
@IsString()
@MinLength(MIN_TICKER_LENGTH)
currency?: string;
@ApiPropertyOptional({ minLength: MIN_TICKER_LENGTH })
@IsOptional()
@IsString()
@MinLength(MIN_TICKER_LENGTH)
realExchange?: string;
@ApiPropertyOptional({ minLength: MIN_TICKER_LENGTH })
@IsOptional()
@IsString()
@MinLength(MIN_TICKER_LENGTH)
countryOfRisk?: string;
@ApiPropertyOptional({ minimum: MIN_TICKER_LENGTH, type: 'integer' })
@IsOptional()
@IsInt()
@Min(MIN_TICKER_LENGTH)
maxLot?: number;
}
export class StockScreenerPageQueryDto {
@ApiProperty({ minimum: MIN_PAGE, type: 'integer' })
@Transform(({ value }: { value: unknown }) => Number(value))
@IsInt()
@Min(MIN_PAGE)
declare page: number;
}
export class StockScreenerItemDto {
@ApiProperty({ format: 'uuid' }) declare instrumentUid: string;
@ApiProperty() declare name: string;
@ApiProperty({ nullable: true, type: String }) declare ticker: null | string;
@ApiProperty({ nullable: true, type: String }) declare isin: null | string;
@ApiProperty({ nullable: true, type: String }) declare logoName: null | string;
@ApiProperty({ nullable: true, type: String }) declare shareType: null | string;
@ApiProperty({ nullable: true, type: String }) declare sector: null | string;
@ApiProperty({ nullable: true, type: String }) declare currency: null | string;
@ApiProperty({ nullable: true, type: String }) declare realExchange: null | string;
@ApiProperty({ nullable: true, type: String }) declare countryOfRisk: null | string;
@ApiProperty({ nullable: true, type: 'integer' }) declare lot: null | number;
}
export class StockScreenerPageDto {
@ApiProperty() declare searchId: string;
@ApiProperty({ format: 'date-time' }) declare asOf: string;
@ApiProperty({ minimum: MIN_PAGE, type: 'integer' }) declare page: number;
@ApiProperty() declare hasNext: boolean;
@ApiProperty({ type: StockScreenerItemDto, isArray: true, maxItems: 30 }) declare items: StockScreenerItemDto[];
}
export class StockSearchQueryDto extends InvestmentReadQueryDto { export class StockSearchQueryDto extends InvestmentReadQueryDto {
@ApiProperty({ minLength: MIN_INSTRUMENT_QUERY_LENGTH, maxLength: MAX_INSTRUMENT_QUERY_LENGTH }) @ApiProperty({ minLength: MIN_INSTRUMENT_QUERY_LENGTH, maxLength: MAX_INSTRUMENT_QUERY_LENGTH })
@@ -209,12 +209,6 @@ export class BrokerageOperationsSyncService {
'tbank.operations.sync.persistence_failed', 'tbank.operations.sync.persistence_failed',
{ userId: attempt.userId, code: TbankErrorCode.Unavailable }, { userId: attempt.userId, code: TbankErrorCode.Unavailable },
); );
// The step could not even record its failure; close the attempt
// durably so it does not stay running. Best effort — an open row
// is re-reserved by the next request.
await this.repository
.fail(attempt, false)
.catch(() => undefined);
} }
const next = pending ? await this.repository.pending(attempt) : null; const next = pending ? await this.repository.pending(attempt) : null;
if (next) { if (next) {
@@ -1,7 +1,6 @@
import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { randomUUID } from 'node:crypto'; import { randomUUID } from 'node:crypto';
import { StructuredLogger } from '../../../common/logging/structured-logger.service';
import { TbankErrorCode } from '../../../integrations/tbank/tbank-error-code'; import { TbankErrorCode } from '../../../integrations/tbank/tbank-error-code';
import { import {
TbankAccessService, TbankAccessService,
@@ -13,9 +12,6 @@ import { BrokerageOperationsSyncService } from '../operations/operations.api';
import { PortfolioService } from '../portfolio/portfolio.api'; import { PortfolioService } from '../portfolio/portfolio.api';
import { UpcomingEventsService } from '../upcoming-events/upcoming-events.service'; import { UpcomingEventsService } from '../upcoming-events/upcoming-events.service';
import { import {
HOURLY_CAUSE_INTERRUPTED,
type HourlyCause,
type HourlyChild,
type HourlySection, type HourlySection,
TbankSyncRepository, TbankSyncRepository,
} from './tbank-sync.repository'; } from './tbank-sync.repository';
@@ -23,19 +19,11 @@ import {
const HOUR_MS = 3_600_000; const HOUR_MS = 3_600_000;
const TICK_MS = 60_000; const TICK_MS = 60_000;
const WORK_POLL_MS = 100; 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 FIRST_DAY_OF_MONTH = 1;
const NO_MILLISECONDS = 0; const ONE_ATTEMPT = 1;
const NO_ATTEMPTS = 0;
type DispatchStep = (child: HourlyChild) => Promise<void>; type SettledWork = { errorCode: null | string; id: string; state: string };
type Accounts = Awaited<ReturnType<TbankAccessService['local']>>['accounts'];
// One native timer per process. The timer only proposes work; the durable // One native timer per process. The timer only proposes work; the durable
// reservation in TbankSyncRepository is the fence that keeps a single // reservation in TbankSyncRepository is the fence that keeps a single
@@ -44,9 +32,6 @@ type Accounts = Awaited<ReturnType<TbankAccessService['local']>>['accounts'];
@Injectable() @Injectable()
export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit { export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
private readonly processRunId = randomUUID(); 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 timer: NodeJS.Timeout | null = null;
private ticking = false; private ticking = false;
@@ -58,7 +43,6 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
private readonly portfolio: PortfolioService, private readonly portfolio: PortfolioService,
private readonly operations: BrokerageOperationsSyncService, private readonly operations: BrokerageOperationsSyncService,
private readonly events: UpcomingEventsService, private readonly events: UpcomingEventsService,
private readonly logger: StructuredLogger,
) {} ) {}
onModuleInit() { onModuleInit() {
@@ -90,7 +74,6 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
// A fresh connection starts its first walk immediately, regardless of the // A fresh connection starts its first walk immediately, regardless of the
// previous term of the replaced connection. // previous term of the replaced connection.
private async launch(userId: string) { private async launch(userId: string) {
await this.recoverAbandoned(userId);
const reserved = await this.repository.reserveHourly( const reserved = await this.repository.reserveHourly(
userId, userId,
this.processRunId, this.processRunId,
@@ -104,307 +87,135 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
return reserved.work; 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 // One membership pass, then the owners refresh their own datasets. The
// walk publishes only safe section results and holds no transaction or // walk publishes only safe section results and holds no transaction or
// upstream call inside one. The attempt outcome is derived from durable // upstream call inside one.
// step rows by the repository, never from walk-local state.
private async walk(attempt: TbankSyncAttempt) { private async walk(attempt: TbankSyncAttempt) {
this.active.add(attempt.attemptId); const failures: HourlySection[] = [];
const attempted = { count: NO_ATTEMPTS };
try { try {
const { userId, revision } = attempt; const { userId, revision } = attempt;
const children: HourlyChild[] = []; await this.refreshMembership(userId, revision, failures, attempted);
// Required sections whose dispatch failed never get a durable step if (failures.some((item) => item.errorCode === TbankErrorCode.InvalidToken)) {
// row; they are recorded as problems so the attempt cannot succeed. return void this.settle(attempt, failures, attempted.count);
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); const current = await this.access.local(userId);
if (current.revision !== revision) {
return void this.settle(attempt, failures, attempted.count);
}
await this.refreshAccounts(userId, revision, current.accounts, failures, attempted);
await this.settle(attempt, failures, attempted.count);
} catch { } catch {
await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED); await this.settle(attempt, failures, attempted.count);
} 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( private async refreshMembership(
attempt: TbankSyncAttempt, userId: string,
dispatch: DispatchStep, revision: string,
dispatchProblems: HourlySection[], failures: HourlySection[],
attempted: { count: number },
) { ) {
const { userId, revision } = attempt; const member = await this.membership.request(userId, revision, true);
const member = await this.withBound( attempted.count += ONE_ATTEMPT;
this.membership.request(userId, revision, true), if (!member) return;
); await member.done;
if (member && member !== BOUND_EXPIRED) { const errorCode = await this.repository.workError(userId, member.id);
await dispatch({ id: member.id, section: 'membership' }); if (errorCode) failures.push({ section: 'membership', errorCode });
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( private async refreshAccounts(
userId: string, userId: string,
revision: string, revision: string,
accounts: Accounts, accounts: Awaited<ReturnType<TbankAccessService['local']>>['accounts'],
dispatch: DispatchStep, failures: HourlySection[],
dispatchProblems: HourlySection[], attempted: { count: number },
) { ) {
for (const account of accounts) { for (const account of accounts) {
await this.dispatchPortfolio( await this.refreshPortfolio(userId, revision, account.accountId, failures, attempted);
userId,
revision,
account.accountId,
dispatch,
dispatchProblems,
);
} }
for (const account of accounts) { for (const account of accounts) {
await this.dispatchOperations( await this.refreshOperations(userId, revision, account.accountId, failures, attempted);
userId,
revision,
account.accountId,
dispatch,
dispatchProblems,
);
} }
for (const account of accounts) { for (const account of accounts) {
await this.dispatchEvents( await this.refreshEvents(userId, account.accountId, failures, attempted);
userId,
account.accountId,
dispatch,
dispatchProblems,
);
} }
} }
private async dispatchPortfolio( private async refreshPortfolio(userId: string, revision: string, accountId: string, failures: HourlySection[], attempted: { count: number }) {
userId: string, const work = await this.portfolio.request(userId, revision, accountId, true);
revision: string, attempted.count += ONE_ATTEMPT;
accountId: string, if (!work) return;
dispatch: DispatchStep, await work.done;
dispatchProblems: HourlySection[], const errorCode = await this.repository.workError(userId, work.id);
) { if (errorCode) failures.push({ section: 'portfolio', errorCode, accountId });
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( private async refreshOperations(userId: string, revision: string, accountId: string, failures: HourlySection[], attempted: { count: number }) {
userId: string, attempted.count += ONE_ATTEMPT;
revision: string,
accountId: string,
dispatch: DispatchStep,
dispatchProblems: HourlySection[],
) {
try { try {
const id = await this.withBound( const id = await this.operations.request(userId, [{ accountId, from: monthStartOf(new Date()), to: new Date() }], true, 'screen', revision);
this.operations.request( if (!id) return;
userId, const errorCode = await this.settledWorkError(userId, [id]);
[ if (errorCode) failures.push({ section: 'operations', errorCode, accountId });
{
accountId,
from: monthStartOf(new Date()),
to: new Date(),
},
],
true,
'screen',
revision,
),
);
if (id && id !== BOUND_EXPIRED) {
await dispatch({ accountId, id, section: 'operations' });
return;
}
} catch { } catch {
// A failed dispatch has no durable step row; record the problem. failures.push({ section: 'operations', errorCode: TbankErrorCode.Unavailable, accountId });
} }
dispatchProblems.push({
accountId,
errorCode: TbankErrorCode.Unavailable,
section: 'operations',
});
} }
private async dispatchEvents( private async refreshEvents(userId: string, accountId: string, failures: HourlySection[], attempted: { count: number }) {
userId: string, attempted.count += ONE_ATTEMPT;
accountId: string,
dispatch: DispatchStep,
dispatchProblems: HourlySection[],
) {
try { try {
const ids = await this.withBound(this.events.refresh(userId, accountId)); const ids = await this.events.refresh(userId, accountId);
if (ids !== BOUND_EXPIRED) { const failed = ids.length ? (await this.settledWorks(userId, ids)).find((row) => row.state === 'failed') : undefined;
for (const id of ids) { if (failed?.errorCode) failures.push({ section: 'events', errorCode: failed.errorCode, accountId });
await dispatch({ accountId, id, section: 'events' }); else await this.events.get(userId, accountId);
}
return;
}
} catch { } catch {
// A failed dispatch has no durable step row; record the problem. failures.push({ section: 'events', errorCode: TbankErrorCode.Unavailable, accountId });
} }
dispatchProblems.push({
accountId,
errorCode: TbankErrorCode.Unavailable,
section: 'events',
});
} }
private async settle( private async settle(
attempt: TbankSyncAttempt, attempt: TbankSyncAttempt,
cause: HourlyCause | null = null, failures: HourlySection[],
dispatchProblems: HourlySection[] = [], attempted: number,
) { ) {
// A suppressed settle error is not lost: the walk leaves the active set let result: 'failed' | 'partial' | 'succeeded' = 'succeeded';
// and the next tick closes the attempt from its durable step results. if (failures.length) {
result = attempted > NO_ATTEMPTS && failures.length >= attempted ? 'failed' : 'partial';
}
await this.repository await this.repository
.settleHourly(attempt, cause, dispatchProblems) .settleHourly(attempt, result, failures)
.catch(() => { .catch(() => undefined);
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 private async settledWorkError(userId: string, ids: string[]) {
// passed the per-step bound. const rows = await this.settledWorks(userId, ids);
private advanceWaiting( return rows.find((row) => row.state === 'failed')?.errorCode ?? null;
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 // Polls the durable work rows until they leave the running state; a row
// bounded so a stuck step cannot hold the attempt running forever; the // deleted with its connection counts as interrupted, not failed.
// bound does not cancel the upstream call, and late results stay in the private async settledWorks(userId: string, ids: string[]) {
// step row where the attempt fence decides whether they count. const settled = new Map<string, SettledWork>();
private async waitWorks(userId: string, ids: string[]) { const pending = new Set(ids);
const waiting = new Map(ids.map((id) => [id, Date.now()])); while (pending.size) {
let pollMs = WORK_POLL_MS; const rows = await this.repository.works(userId, [...pending]);
while (waiting.size) { for (const row of rows) {
const rows = await this.repository.works(userId, [...waiting.keys()]); if (row.state !== 'running') {
this.advanceWaiting(waiting, rows); settled.set(row.id, row);
if (waiting.size) { pending.delete(row.id);
await new Promise((resolve) => setTimeout(resolve, pollMs)); }
pollMs = Math.min(pollMs * WORK_POLL_BACKOFF_FACTOR, WORK_POLL_MAX_MS);
} }
for (const id of [...pending]) {
if (!rows.some((row) => row.id === id)) {
settled.set(id, { id, state: 'cancelled', errorCode: null });
pending.delete(id);
}
}
if (pending.size) await new Promise((resolve) => setTimeout(resolve, WORK_POLL_MS));
} }
return [...settled.values()];
} }
} }
@@ -18,19 +18,7 @@ export type HourlySection = {
errorCode: null | string; errorCode: null | string;
section: 'events' | 'membership' | 'operations' | 'portfolio'; section: 'events' | 'membership' | 'operations' | 'portfolio';
}; };
export type HourlyChild = {
accountId?: string;
id: string;
section: HourlySection['section'];
};
export const HOURLY_CAUSE_DEADLINE = 'DEADLINE';
export const HOURLY_CAUSE_INTERRUPTED = 'INTERRUPTED';
export type HourlyCause = typeof HOURLY_CAUSE_INTERRUPTED;
type StepOutcome = { problem?: HourlySection; success: boolean };
type StepRow = { attemptId: string; errorCode: null | string; state: string };
export type HourlyProgress = { export type HourlyProgress = {
attempts?: Record<string, string>;
children?: HourlyChild[];
result?: 'failed' | 'partial' | 'succeeded'; result?: 'failed' | 'partial' | 'succeeded';
sections: HourlySection[]; sections: HourlySection[];
}; };
@@ -203,10 +191,12 @@ export class TbankSyncRepository {
const due = now.valueOf() - previous.createdAt.valueOf() >= hourMs; const due = now.valueOf() - previous.createdAt.valueOf() >= hourMs;
if (previous.state === 'running') { if (previous.state === 'running') {
if (previous.processRunId === runId) return null; if (previous.processRunId === runId) return null;
// A dead process cannot finish its attempt; close it from its // A dead process cannot finish its attempt; interrupt it, but its
// actual step results, but its start stays the basis of the next // start stays the basis of the next term.
// term. await tx.tbankSyncWork.update({
await this.closeHourly(tx, previous, 'INTERRUPTED', []); where: { userId_revision_scopeKey: key },
data: { state: 'cancelled', errorCode: null },
});
if (!due) return null; if (!due) return null;
} else if (!due) return null; } else if (!due) return null;
} }
@@ -230,146 +220,28 @@ export class TbankSyncRepository {
}); });
} }
private classifyStep( settleHourly(
child: HourlyChild, attempt: TbankSyncAttempt,
row: StepRow | undefined, result: 'failed' | 'partial' | 'succeeded',
recorded: string | undefined,
cause: HourlyCause | null,
): StepOutcome {
if (row && row.attemptId === recorded && row.state === 'complete') {
return { success: true };
}
const stale = !row || (recorded !== undefined && row.attemptId !== recorded);
let errorCode: null | string = null;
if (!stale && row.state === 'failed') errorCode = row.errorCode;
else if (!stale && cause === HOURLY_CAUSE_INTERRUPTED) {
errorCode = HOURLY_CAUSE_INTERRUPTED;
} else if (!stale) errorCode = HOURLY_CAUSE_DEADLINE;
return {
problem: { accountId: child.accountId, errorCode, section: child.section },
success: false,
};
}
private deriveResult(
cause: HourlyCause | null,
sections: HourlySection[], sections: HourlySection[],
successes: number,
): 'failed' | 'partial' | 'succeeded' {
if (!cause && !sections.length) return 'succeeded';
return successes ? 'partial' : 'failed';
}
// The outcome is derived from the durable state of the dispatched steps,
// never from an in-memory error list that an exception path can empty.
// `cause` marks a close before normal completion and caps the result at
// `partial`; `dispatchProblems` records required sections that never got a
// durable step row because their dispatch failed.
private async closeHourly(
tx: Prisma.TransactionClient,
work: {
attemptId: string;
id: string;
progress: Prisma.JsonValue;
revision: string;
userId: string;
},
cause: HourlyCause | null,
dispatchProblems: HourlySection[],
) { ) {
const progress = work.progress as HourlyProgress; return this.prisma.tbankSyncWork.updateMany({
const children = progress.children ?? []; where: {
const rows = children.length id: attempt.id,
? await tx.tbankSyncWork.findMany({ attemptId: attempt.attemptId,
where: { userId: attempt.userId,
userId: work.userId, revision: attempt.revision,
id: { in: children.map((child) => child.id) }, state: 'running',
}, },
select: { attemptId: true, errorCode: true, id: true, state: true },
})
: [];
const byId = new Map(rows.map((row) => [row.id, row]));
const outcomes = children.map((child) =>
this.classifyStep(
child,
byId.get(child.id),
progress.attempts?.[child.id],
cause,
),
);
const sections = outcomes
.flatMap((outcome) => (outcome.problem ? [outcome.problem] : []))
.concat(dispatchProblems);
const successes = outcomes.filter((outcome) => outcome.success).length;
const result = this.deriveResult(cause, sections, successes);
return tx.tbankSyncWork.updateMany({
where: { id: work.id, attemptId: work.attemptId, state: 'running' },
data: { data: {
state: result === 'failed' ? 'failed' : 'complete', state: result === 'failed' ? 'failed' : 'complete',
errorCode: errorCode:
cause ?? sections.find((section) => section.errorCode)?.errorCode ?? null,
sections.find((section) => section.errorCode)?.errorCode ?? progress: { result, sections },
null,
progress: {
attempts: progress.attempts ?? {},
children,
result,
sections,
},
}, },
}); });
} }
async settleHourly(
attempt: TbankSyncAttempt,
cause: HourlyCause | null = null,
dispatchProblems: HourlySection[] = [],
) {
return this.prisma.$transaction(async (tx) => {
const work = await tx.tbankSyncWork.findFirst({ where: attempt });
if (!work || work.state !== 'running') return { count: 0 };
return this.closeHourly(tx, work, cause, dispatchProblems);
});
}
// Records a dispatched step on the attempt before it is awaited, so a lost
// walk can still derive its outcome from durable rows.
async recordHourlyChild(attempt: TbankSyncAttempt, child: HourlyChild) {
return this.prisma.$transaction(async (tx) => {
const work = await tx.tbankSyncWork.findFirst({
where: { ...attempt, state: 'running' },
select: { progress: true },
});
if (!work) return false;
const progress = work.progress as HourlyProgress;
const row = await tx.tbankSyncWork.findFirst({
where: { userId: attempt.userId, id: child.id },
select: { attemptId: true },
});
const updated = await tx.tbankSyncWork.updateMany({
where: { ...attempt, state: 'running' },
data: {
progress: {
attempts: {
...progress.attempts,
...(row ? { [child.id]: row.attemptId } : {}),
},
children: [...(progress.children ?? []), child],
sections: progress.sections ?? [],
},
},
});
return Boolean(updated.count);
});
}
runningHourly(userId: string) {
return this.prisma.tbankSyncWork.findMany({
where: { userId, scopeKey: 'hourly', state: 'running' },
select: { attemptId: true, id: true, processRunId: true, revision: true },
});
}
async hourly(userId: string) { async hourly(userId: string) {
const connection = await this.prisma.tbankConnection.findFirst({ const connection = await this.prisma.tbankConnection.findFirst({
where: { userId }, where: { userId },
-47
View File
@@ -15,7 +15,6 @@ import { configureApp } from '../src/main';
import { StockCatalogService } from '../src/modules/investments/instruments/stocks/application/stock-catalog.service'; import { StockCatalogService } from '../src/modules/investments/instruments/stocks/application/stock-catalog.service';
import { StockIncomeService } from '../src/modules/investments/instruments/stocks/application/stock-income.service'; import { StockIncomeService } from '../src/modules/investments/instruments/stocks/application/stock-income.service';
import { StockMarketService } from '../src/modules/investments/instruments/stocks/application/stock-market.service'; import { StockMarketService } from '../src/modules/investments/instruments/stocks/application/stock-market.service';
import { StockScreenerService } from '../src/modules/investments/instruments/stocks/application/stock-screener.service';
const uid = 'e6123145-9665-43e0-8413-cd61b8aa9b13'; const uid = 'e6123145-9665-43e0-8413-cd61b8aa9b13';
describe('Stock HTTP contract', () => { describe('Stock HTTP contract', () => {
@@ -26,7 +25,6 @@ describe('Stock HTTP contract', () => {
search: jest.fn(), search: jest.fn(),
candles: jest.fn(), candles: jest.fn(),
}; };
const screener = { page: jest.fn(), search: jest.fn() };
const token = new JwtService().sign( const token = new JwtService().sign(
{ sub: 'stock-reader', type: 'access' }, { sub: 'stock-reader', type: 'access' },
{ secret: process.env.JWT_SECRET }, { secret: process.env.JWT_SECRET },
@@ -39,8 +37,6 @@ describe('Stock HTTP contract', () => {
.useValue(stocks) .useValue(stocks)
.overrideProvider(StockIncomeService) .overrideProvider(StockIncomeService)
.useValue(stocks) .useValue(stocks)
.overrideProvider(StockScreenerService)
.useValue(screener)
.compile(); .compile();
app = module.createNestApplication(); app = module.createNestApplication();
configureApp(app); configureApp(app);
@@ -55,55 +51,12 @@ describe('Stock HTTP contract', () => {
}); });
stocks.search.mockResolvedValue([]); stocks.search.mockResolvedValue([]);
stocks.getByTicker.mockResolvedValue({ instrumentUid: uid, slug: 'SBERP' }); stocks.getByTicker.mockResolvedValue({ instrumentUid: uid, slug: 'SBERP' });
screener.search.mockResolvedValue({ asOf: '2026-10-10T12:00:00.000Z', hasNext: false, items: [], page: 1, searchId: 'search-id' });
}); });
afterAll(() => app.close()); afterAll(() => app.close());
it('requires authentication before invoking the use case', async () => { it('requires authentication before invoking the use case', async () => {
await request(app.getHttpServer()).get(`/api/v1/stocks/${uid}`).expect(401); await request(app.getHttpServer()).get(`/api/v1/stocks/${uid}`).expect(401);
expect(stocks.get).not.toHaveBeenCalled(); expect(stocks.get).not.toHaveBeenCalled();
}); });
it.each([
[() => request(app.getHttpServer()).post('/api/v1/stocks/screener/search'), 'post'],
[() => request(app.getHttpServer()).get('/api/v1/stocks/screener/search/search-id?page=1'), 'get'],
])('requires authentication for screener route', async (send) => {
await send().expect(401);
expect(screener.search).not.toHaveBeenCalled();
expect(screener.page).not.toHaveBeenCalled();
});
it('validates screener input and returns raw snapshot pages', async () => {
await request(app.getHttpServer())
.post('/api/v1/stocks/screener/search')
.set('Authorization', `Bearer ${token}`)
.send({ query: ' ', unknown: 'value' })
.expect(400);
expect(screener.search).not.toHaveBeenCalled();
const response = await request(app.getHttpServer())
.post('/api/v1/stocks/screener/search')
.set('Authorization', `Bearer ${token}`)
.send({ query: '' })
.expect(200);
expect(response.body).toEqual({ asOf: '2026-10-10T12:00:00.000Z', hasNext: false, items: [], page: 1, searchId: 'search-id' });
expect(screener.search).toHaveBeenCalledWith('stock-reader', {});
});
it.each([
'/api/v1/stocks/screener/search/search-id',
'/api/v1/stocks/screener/search/search-id?page=0',
'/api/v1/stocks/screener/search/search-id?page=one',
])('validates required screener page on %s', async (path) => {
await request(app.getHttpServer())
.get(path)
.set('Authorization', `Bearer ${token}`)
.expect(400);
expect(screener.page).not.toHaveBeenCalled();
});
it('reads a screener snapshot page with the verified identity', async () => {
await request(app.getHttpServer())
.get('/api/v1/stocks/screener/search/search-id?page=2')
.set('Authorization', `Bearer ${token}`)
.expect(200);
expect(screener.page).toHaveBeenCalledWith('stock-reader', 'search-id', 2);
});
it.each([ it.each([
'/stocks/not-a-uid', '/stocks/not-a-uid',
'/stocks/by-ticker/%20', '/stocks/by-ticker/%20',
-582
View File
@@ -1,582 +0,0 @@
import { ConfigService } from '@nestjs/config';
import assert from 'node:assert/strict';
import { randomUUID } from 'node:crypto';
import { setImmediate as immediate } from 'node:timers/promises';
import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service';
import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
import { CatalogEnrichmentService } from '../src/modules/investments/instruments/application/catalog-enrichment.service';
import { CatalogDiscoveryRepository } from '../src/modules/investments/instruments/infrastructure/catalog-discovery.repository';
import { BrokerageOperationsSyncService } from '../src/modules/investments/operations/application/brokerage-operations-sync.service';
import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository';
import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler';
import { TbankSyncRepository } from '../src/modules/investments/tbank-sync/tbank-sync.repository';
import { PrismaService } from '../src/prisma/prisma.service';
type HourlyProgressView = {
result?: string;
sections: { accountId?: string; errorCode: null | string; section: string }[];
};
// Real SQLite proves reservation, outcome derivation, recovery and fences;
// the step owners are controlled so each failure mode can be injected.
describe('hourly attempt lifecycle (SQLite)', () => {
const prisma = new PrismaService(
new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
);
const tokens = new TbankTokenService(
new ConfigService({
TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY,
}) as never,
);
const repository = new TbankSyncRepository(prisma);
const access = { local: jest.fn() };
const membership = { request: jest.fn() };
const portfolio = { request: jest.fn() };
const operations = { request: jest.fn() };
const events = { refresh: jest.fn(), get: jest.fn() };
const logger = { event: jest.fn() };
const createScheduler = () =>
new TbankHourlyScheduler(
prisma,
repository,
access as never,
membership as never,
portfolio as never,
operations as never,
events as never,
logger as never,
);
const USER = 'hourly-owner';
const REVISION = 'hourly-revision';
function child(
state: 'complete' | 'failed' | 'running',
errorCode: null | string,
scopeKey: string,
) {
return prisma.tbankSyncWork.create({
data: {
id: randomUUID(),
userId: USER,
revision: REVISION,
scopeKey: `test:${scopeKey}:${randomUUID()}`,
attemptId: randomUUID(),
processRunId: 'child-process',
priority: 0,
intent: {},
progress: {},
state,
errorCode,
},
});
}
function hourlyRow() {
return prisma.tbankSyncWork.findFirst({
where: { userId: USER, scopeKey: 'hourly' },
});
}
async function pump(turns = 20) {
for (let i = 0; i < turns; i++) {
await immediate();
await jest.advanceTimersByTimeAsync(50);
}
}
async function runWalk(scheduler: TbankHourlyScheduler, advanceMs = 50) {
await scheduler.tick();
for (let i = 0; i < 400; i++) {
await immediate();
await jest.advanceTimersByTimeAsync(advanceMs);
const row = await hourlyRow();
if (row && row.state !== 'running') return row;
}
throw new Error('hourly attempt did not settle');
}
// Sequential per-step bounds: advance past each bound until the attempt
// closes. Covers walks whose awaits hang before any durable step row exists.
async function runBoundedWalk(scheduler: TbankHourlyScheduler) {
await scheduler.tick();
for (let i = 0; i < 12; i++) {
await pump(5);
await jest.advanceTimersByTimeAsync(16 * 60 * 1000);
const row = await hourlyRow();
if (row && row.state !== 'running') return row;
}
throw new Error('hourly attempt did not settle');
}
beforeAll(async () => {
await prisma.$connect();
await prisma.user.create({
data: { id: USER, email: 'hourly@example.invalid', passwordHash: 'x' },
});
});
afterAll(async () => prisma.$disconnect());
beforeEach(async () => {
jest.resetAllMocks();
await prisma.tbankSyncWork.deleteMany();
await prisma.tbankConnection.deleteMany();
await prisma.tbankConnection.create({
data: {
userId: USER,
revision: REVISION,
encryptedToken: tokens.encrypt('synthetic'),
connectedAt: new Date(),
},
});
jest.useFakeTimers({
doNotFake: ['queueMicrotask', 'setImmediate', 'nextTick'],
});
access.local.mockResolvedValue({
revision: REVISION,
accounts: [{ accountId: 'a' }, { accountId: 'b' }],
});
membership.request.mockImplementation(async () => {
const row = await child('complete', null, 'membership');
return { id: row.id, done: Promise.resolve() };
});
portfolio.request.mockImplementation(
async (_userId: string, _revision: string, accountId: string) => {
const row = await child('complete', null, `portfolio:${accountId}`);
return { id: row.id, done: Promise.resolve() };
},
);
operations.request.mockImplementation(async () => {
const row = await child('complete', null, 'operations');
return row.id;
});
events.refresh.mockResolvedValue([]);
events.get.mockResolvedValue(undefined);
});
afterEach(() => {
jest.useRealTimers();
jest.restoreAllMocks();
});
it('settles the first walk succeeded only when every step row completed', async () => {
const row = await runWalk(createScheduler());
expect(row.state).toBe('complete');
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('succeeded');
expect(progress.sections).toEqual([]);
const read = await repository.hourly(USER);
expect(read?.state).toBe('succeeded');
});
it('settles partial and keeps successful steps when one account fails', async () => {
portfolio.request.mockImplementation(
async (_userId: string, _revision: string, accountId: string) => {
const failed = accountId === 'b';
const row = await child(
failed ? 'failed' : 'complete',
failed ? 'TBANK_UNAVAILABLE' : null,
`portfolio:${accountId}`,
);
return { id: row.id, done: Promise.resolve() };
},
);
const row = await runWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('partial');
expect(progress.sections).toHaveLength(1);
expect(progress.sections[0].section).toBe('portfolio');
expect(progress.sections[0].accountId).toBe('b');
expect(progress.sections[0].errorCode).toBe('TBANK_UNAVAILABLE');
});
it('settles failed when the membership step fails before any update', async () => {
membership.request.mockImplementation(async () => {
const row = await child('failed', 'INVALID_TBANK_TOKEN', 'membership');
return { id: row.id, done: Promise.resolve() };
});
const row = await runWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('failed');
expect(row.errorCode).toBe('INVALID_TBANK_TOKEN');
expect(progress.sections[0].section).toBe('membership');
});
it('never succeeds when no required step succeeded', async () => {
membership.request.mockImplementation(async () => {
const row = await child('failed', 'TBANK_UNAVAILABLE', 'membership');
return { id: row.id, done: Promise.resolve() };
});
portfolio.request.mockImplementation(
async (_userId: string, _revision: string, accountId: string) => {
const row = await child(
'failed',
'TBANK_UNAVAILABLE',
`portfolio:${accountId}`,
);
return { id: row.id, done: Promise.resolve() };
},
);
operations.request.mockRejectedValue(new Error('db'));
events.refresh.mockRejectedValue(new Error('db'));
const row = await runWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('failed');
expect(progress.result).not.toBe('succeeded');
assert.ok(progress.sections.length > 0);
});
it('caps the outcome at partial when the coordinator itself breaks', async () => {
access.local.mockRejectedValueOnce(new Error('db'));
const row = await runWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('partial');
expect(row.errorCode).toBe('INTERRUPTED');
});
it('records a section whose dispatch failed as a problem', async () => {
events.refresh.mockRejectedValue(new Error('db'));
const row = await runWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('partial');
const eventsSection = progress.sections.find(
(entry) => entry.section === 'events',
);
assert.ok(eventsSection);
expect(eventsSection.errorCode).toBe('TBANK_UNAVAILABLE');
expect(row.errorCode).toBe('TBANK_UNAVAILABLE');
});
it('closes a stuck step as DEADLINE and keeps its row running', async () => {
operations.request.mockImplementation(async () => {
const row = await child('running', null, 'operations');
return row.id;
});
const scheduler = createScheduler();
await scheduler.tick();
await pump();
await jest.advanceTimersByTimeAsync(16 * 60 * 1000);
await pump(20);
const row = await hourlyRow();
assert.ok(row);
expect(row.state).not.toBe('running');
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('partial');
assert.ok(
progress.sections.some(
(entry) =>
entry.section === 'operations' && entry.errorCode === 'DEADLINE',
),
);
const stuck = await prisma.tbankSyncWork.findFirst({
where: { userId: USER, state: 'running' },
});
assert.ok(stuck, 'the wait bound does not cancel the upstream call');
});
it('settles the attempt when a portfolio dispatch never resolves', async () => {
portfolio.request.mockImplementation(() => new Promise(() => undefined));
const row = await runBoundedWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('partial');
const problems = progress.sections.filter(
(entry) => entry.section === 'portfolio',
);
expect(problems).toHaveLength(2);
expect(problems[0].errorCode).toBe('TBANK_UNAVAILABLE');
});
it('settles the attempt when the membership dispatch never resolves', async () => {
membership.request.mockImplementation(() => new Promise(() => undefined));
const row = await runBoundedWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('partial');
expect(progress.sections).toHaveLength(1);
expect(progress.sections[0].section).toBe('membership');
expect(progress.sections[0].errorCode).toBe('TBANK_UNAVAILABLE');
});
it('bounds a membership step promise that never resolves', async () => {
membership.request.mockImplementation(async () => {
const row = await child('running', null, 'membership');
return { id: row.id, done: new Promise(() => undefined) };
});
const row = await runBoundedWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('partial');
const membershipSection = progress.sections.find(
(entry) => entry.section === 'membership',
);
expect(membershipSection?.errorCode).toBe('DEADLINE');
});
it('settles the attempt when a post-wait refresh never resolves', async () => {
events.get.mockImplementation(() => new Promise(() => undefined));
const row = await runBoundedWalk(createScheduler());
const progress = row.progress as HourlyProgressView;
expect(progress.result).toBe('succeeded');
expect(progress.sections).toEqual([]);
});
it('recovers a lost settle and unblocks the next attempt at its term', async () => {
const settleSpy = jest
.spyOn(repository, 'settleHourly')
.mockRejectedValueOnce(new Error('db'));
const scheduler = createScheduler();
await scheduler.tick();
for (let i = 0; i < 40 && settleSpy.mock.calls.length === 0; i++) {
await immediate();
await jest.advanceTimersByTimeAsync(50);
}
expect(settleSpy.mock.calls).toHaveLength(1);
const open = await hourlyRow();
expect(open?.state).toBe('running');
await pump(20);
await scheduler.tick();
await pump(20);
const closed = await hourlyRow();
expect(closed?.state).toBe('complete');
expect((closed?.progress as HourlyProgressView).result).toBe('partial');
expect(closed?.errorCode).toBe('INTERRUPTED');
await jest.advanceTimersByTimeAsync(61 * 60 * 1000);
await scheduler.tick();
await pump(30);
const next = await hourlyRow();
expect(next?.attemptId).not.toBe(closed?.attemptId);
expect(next?.state).not.toBe('running');
expect((next?.progress as HourlyProgressView).result).toBe('succeeded');
});
it('closes an abandoned foreign-process attempt and keeps the normal term', async () => {
const stuck = await child('complete', null, 'membership');
const deadAttemptId = randomUUID();
await prisma.tbankSyncWork.create({
data: {
id: randomUUID(),
userId: USER,
revision: REVISION,
scopeKey: 'hourly',
attemptId: deadAttemptId,
processRunId: 'dead-process',
priority: 0,
intent: {},
state: 'running',
progress: {
children: [{ id: stuck.id, section: 'membership' }],
attempts: { [stuck.id]: stuck.attemptId },
sections: [],
},
},
});
const scheduler = createScheduler();
await runWalk(scheduler);
const closed = await hourlyRow();
expect(closed?.attemptId).toBe(deadAttemptId);
expect(closed?.state).not.toBe('running');
expect((closed?.progress as HourlyProgressView).result).toBe('partial');
expect(closed?.errorCode).toBe('INTERRUPTED');
const hourlyCount = await prisma.tbankSyncWork.count({
where: { userId: USER, scopeKey: 'hourly' },
});
expect(hourlyCount).toBe(1);
await jest.advanceTimersByTimeAsync(61 * 60 * 1000);
await scheduler.tick();
await pump(30);
const next = await hourlyRow();
expect(next?.attemptId).not.toBe(deadAttemptId);
expect(next?.state).not.toBe('running');
});
it('does not overlap a running attempt with a second walk', async () => {
operations.request.mockImplementation(async () => {
const row = await child('running', null, 'operations');
return row.id;
});
const scheduler = createScheduler();
await scheduler.tick();
await pump();
expect((await hourlyRow())?.state).toBe('running');
await scheduler.tick();
await pump();
const hourlyCount = await prisma.tbankSyncWork.count({
where: { userId: USER, scopeKey: 'hourly' },
});
expect(hourlyCount).toBe(1);
expect((await hourlyRow())?.state).toBe('running');
await jest.advanceTimersByTimeAsync(16 * 60 * 1000);
await pump(20);
expect((await hourlyRow())?.state).not.toBe('running');
});
it('does not start a new attempt before the next hourly term', async () => {
const scheduler = createScheduler();
const row = await runWalk(scheduler);
const firstAttemptId = row.attemptId;
expect(row.state).toBe('complete');
await scheduler.tick();
expect((await hourlyRow())?.attemptId).toBe(firstAttemptId);
await jest.advanceTimersByTimeAsync(59 * 60 * 1000);
await scheduler.tick();
expect((await hourlyRow())?.attemptId).toBe(firstAttemptId);
await jest.advanceTimersByTimeAsync(2 * 60 * 1000);
await scheduler.tick();
await pump(30);
expect((await hourlyRow())?.attemptId).not.toBe(firstAttemptId);
});
it('counts a late step result before the attempt outcome is recorded', async () => {
const late: { id: string }[] = [];
operations.request.mockImplementation(async () => {
const row = await child('running', null, 'operations');
late.push(row);
return row.id;
});
const scheduler = createScheduler();
await scheduler.tick();
await pump();
assert.ok(late.length > 0);
for (const row of late) {
await prisma.tbankSyncWork.update({
where: { id: row.id },
data: { state: 'complete' },
});
}
let settled = false;
for (let i = 0; i < 100 && !settled; i++) {
await immediate();
await jest.advanceTimersByTimeAsync(200);
const current = await hourlyRow();
settled = Boolean(current && current.state !== 'running');
}
const row = await hourlyRow();
expect(row).not.toBeNull();
expect(row?.state).not.toBe('running');
expect((row?.progress as HourlyProgressView).result).toBe('succeeded');
});
it('treats a step superseded by a successor as stale and keeps CAS fencing', async () => {
const stale = await child('complete', null, 'membership');
const attemptId = randomUUID();
const id = randomUUID();
await prisma.tbankSyncWork.create({
data: {
id,
userId: USER,
revision: REVISION,
scopeKey: 'hourly',
attemptId,
processRunId: 'dead-process',
priority: 0,
intent: {},
state: 'running',
progress: {
children: [{ id: stale.id, section: 'membership' }],
attempts: { [stale.id]: randomUUID() },
sections: [],
},
},
});
const foreign = await repository.settleHourly(
{ userId: USER, revision: REVISION, id, attemptId: randomUUID() },
'INTERRUPTED',
);
expect(foreign.count).toBe(0);
expect((await hourlyRow())?.state).toBe('running');
const closed = await repository.settleHourly(
{ userId: USER, revision: REVISION, id, attemptId },
'INTERRUPTED',
);
expect(closed.count).toBe(1);
const progress = (await hourlyRow())?.progress as HourlyProgressView;
expect(progress.result).toBe('failed');
expect(progress.sections[0].errorCode).toBeNull();
const late = await repository.settleHourly(
{ userId: USER, revision: REVISION, id, attemptId },
'INTERRUPTED',
);
expect(late.count).toBe(0);
});
it('creates no reservations after a controlled shutdown', async () => {
const scheduler = createScheduler();
scheduler.onModuleInit();
scheduler.onModuleDestroy();
await jest.advanceTimersByTimeAsync(2 * 60 * 60 * 1000);
await pump(5);
expect(await prisma.tbankSyncWork.count({ where: { userId: USER } })).toBe(0);
});
it('closes the operations step durably when its finalizer cannot record', async () => {
await prisma.tbankConnection.update({
where: { userId: USER },
data: { membershipAt: new Date() },
});
await prisma.tbankAccount.create({
data: {
userId: USER,
revision: REVISION,
accountId: 'a',
type: 'BROKER',
name: 'Test',
upstreamOrder: 0,
},
});
const accessReal = new TbankAccessService(prisma, tokens);
const operationsRepository = new BrokerageOperationsRepository(prisma);
const sync = new BrokerageOperationsSyncService(
operationsRepository,
accessReal,
{ getOperationsByCursor: jest.fn() } as never,
logger as never,
);
jest
.spyOn(operationsRepository, 'pending')
.mockRejectedValueOnce(new Error('db'));
jest
.spyOn(operationsRepository, 'fail')
.mockRejectedValueOnce(new Error('db'));
const id = await sync.request(
USER,
[
{
accountId: 'a',
from: new Date(Date.UTC(2026, 0, 1)),
to: new Date(),
},
] as never,
true,
'screen',
REVISION,
);
assert.ok(id);
await pump(40);
const row = await prisma.tbankSyncWork.findFirst({
where: { userId: USER, id },
});
expect(row?.state).toBe('failed');
});
it('finishes open catalog steps durably when publication fails mid-batch', async () => {
const accessReal = new TbankAccessService(prisma, tokens);
const catalogRepository = new CatalogDiscoveryRepository(prisma);
const enrichment = new CatalogEnrichmentService(
catalogRepository,
accessReal,
{ getLastPrices: jest.fn().mockResolvedValue([]) } as never,
logger as never,
{} as never,
{} as never,
{} as never,
);
jest
.spyOn(catalogRepository, 'publishSource')
.mockRejectedValue(new Error('db'));
await enrichment.requestMany(USER, REVISION, 'share', ['uid'], false, [
'last-price',
]);
await pump(40);
const rows = await prisma.tbankSyncWork.findMany({
where: { userId: USER, scopeKey: { not: 'hourly' } },
});
assert.ok(rows.length > 0);
for (const row of rows) expect(row.state).toBe('failed');
});
});