Compare commits

...

9 Commits

Author SHA1 Message Date
dependabot[bot] 7270d5656a chore(deps): Bump react and @types/react in /web
Bumps [react](https://github.com/facebook/react/tree/HEAD/packages/react) and [@types/react](https://github.com/DefinitelyTyped/DefinitelyTyped/tree/HEAD/types/react). These dependencies needed to be updated together.

Updates `react` from 18.3.1 to 19.2.7
- [Release notes](https://github.com/facebook/react/releases)
- [Changelog](https://github.com/react/react/blob/main/CHANGELOG.md)
- [Commits](https://github.com/facebook/react/commits/v19.2.7/packages/react)

Updates `@types/react` from 18.3.31 to 19.2.17
- [Release notes](https://github.com/DefinitelyTyped/DefinitelyTyped/releases)
- [Commits](https://github.com/DefinitelyTyped/DefinitelyTyped/commits/HEAD/types/react)

---
updated-dependencies:
- dependency-name: "@types/react"
  dependency-version: 19.2.17
  dependency-type: direct:development
  update-type: version-update:semver-major
- dependency-name: react
  dependency-version: 19.2.7
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-07-16 19:01:24 +00:00
Mauricio Barragan 02a5cbe430 Merge pull request #3 from mauricioabh/dev
Production Skills M0–M5: OpenAPI, Sentry, OTel, Upstash
2026-07-16 12:59:03 -06:00
Mauricio Barragan a5c83ee026 feat(observability): gate Sentry tracing spans behind SENTRY_TRACING
Spans are now created and exported to Sentry only when SENTRY_TRACING is enabled (default off), keeping 24/7 operation within the free-tier span quota. Error monitoring stays always-on. Includes OpenSpec change gate-sentry-spans (proposal, design, specs, tasks).

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-16 12:54:15 -06:00
Mauricio Barragan f6e8381c20 test(m3): add Upstash read-tier rate limit HTTP test
Node test asserts 429 with Retry-After after 60 GET requests per IP.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-17 21:35:11 -06:00
Mauricio Barragan 8c7127cfde Add Sentry dev probe and observability smoke test.
GET /api/debug/sentry, load .env.local for DSN, Playwright observability spec.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-12 10:38:08 -06:00
Mauricio Barragan e04f3977c6 Fix Upstash import path blocking server start (M5 QA).
rate-limit.ts imported from a non-existent interfaces/cache path; use infrastructure/cache so Playwright smoke can start the server.
2026-06-11 03:06:42 -06:00
Mauricio Barragan fd4170963e Add Sentry, Pino logs, and OTel spans on WS pipeline (M2).
Capture REST/WS/SSE errors in Sentry, emit structured pino logs with correlation IDs, and trace the book-tick pipeline to Sentry via OpenTelemetry.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-11 01:01:20 -06:00
Mauricio Barragan f4def0c214 Refactor API contracts to Zod, zod-to-openapi, and Scalar.
Replace hand-written OpenAPI and Swagger UI with schema-driven docs at /api-docs and Zod validation on request bodies.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-11 00:04:12 -06:00
Mauricio Barragan b5ff33d716 Add Husky pre-commit with eslint and Sentry env convention (M0).
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-10 23:51:52 -06:00
42 changed files with 4530 additions and 876 deletions
+15
View File
@@ -22,3 +22,18 @@ DEMO_MODE=false
# Feed recorder (NDJSON to data/)
RECORD_FEED=false
# API docs — Scalar at GET /api-docs (OpenAPI generated from Zod; no env vars required)
# Observability — https://sentry.io (project: arbpulse)
SENTRY_DSN=
# Tracing/spans gate — default off. Error monitoring is always on when SENTRY_DSN
# is set; spans are only created/exported when SENTRY_TRACING is enabled. Kept off
# by default so 24/7 operation stays within Sentry's free-tier span quota.
SENTRY_TRACING=false
# Structured logs (pino); correlation IDs on REST, SSE, and WS hot path
LOG_LEVEL=info
# Upstash Redis — rate limit on /api/* and short-TTL snapshot cache for GET /api/state
UPSTASH_REDIS_REST_URL="https://natural-baboon-144595.upstash.io"
UPSTASH_REDIS_REST_TOKEN="gQAAAAAAAjTTAAIgcDE0MDk4MGZlOWMwN2M0NTdiOWQ5MWYxMjVhOWMyNWVlZQ"
+30
View File
@@ -0,0 +1,30 @@
name: CodeQL Security Scan
on:
push:
branches: [main, dev]
pull_request:
branches: [main, dev]
schedule:
- cron: "0 8 * * 1"
jobs:
analyze:
name: Analyze
runs-on: ubuntu-latest
permissions:
actions: read
contents: read
security-events: write
steps:
- uses: actions/checkout@v4
- uses: github/codeql-action/init@v3
with:
languages: javascript-typescript
queries: security-extended
- uses: github/codeql-action/autobuild@v3
- uses: github/codeql-action/analyze@v3
+45
View File
@@ -0,0 +1,45 @@
name: E2E Smoke
on:
push:
branches: [main, dev]
pull_request:
branches: [main, dev]
jobs:
smoke:
name: Playwright dashboard + SSE
runs-on: ubuntu-latest
env:
DEMO_MODE: "true"
steps:
- uses: actions/checkout@v4
- uses: actions/setup-node@v4
with:
node-version: "20"
cache: npm
cache-dependency-path: |
package-lock.json
web/package-lock.json
- run: npm ci
- run: npm --prefix web ci
- run: npm run build
- name: Start server
run: npm start &
- name: Wait for API
run: |
for i in $(seq 1 45); do
curl -sf http://localhost:8080/api/health > /dev/null && echo "Ready after ${i}s" && exit 0
sleep 1
done
echo "::error::Server did not become ready in 45s" && exit 1
- run: npx playwright install --with-deps chromium
- name: Run smoke tests
run: npm run test:e2e
+1
View File
@@ -0,0 +1 @@
npx lint-staged
+6
View File
@@ -0,0 +1,6 @@
{
"semi": true,
"singleQuote": false,
"trailingComma": "all",
"tabWidth": 2
}
+13 -2
View File
@@ -301,9 +301,9 @@ Respuestas REST siguen la forma `{ success, data?, error? }`.
| `POST` | `/api/control/record` | `{ "enabled": boolean }` — grabación NDJSON en `data/` |
| `POST` | `/api/control/threshold` | `{ "pct": number }` (atajo de `PATCH /api/config`) |
| `POST` | `/api/control/max-trade` | `{ "btc": number }` (atajo de `PATCH /api/config`) |
| `GET` | `/api-docs` | **Swagger UI** — documentación interactiva OpenAPI 3.0 |
| `GET` | `/api-docs` | **Scalar** — documentación interactiva OpenAPI 3.0 |
Contrato completo y schemas en `src/interfaces/http/openapi.ts`.
Contrato generado desde Zod (`src/interfaces/http/schemas/`) con `@asteasolutions/zod-to-openapi`; registro de rutas en `src/interfaces/http/openapi.ts`.
---
@@ -343,6 +343,17 @@ Eso no indica que el bot “no funcione”: indica que el filtro económico es e
---
## Production practices
- **Pre-commit:** Husky runs lint-staged (`eslint --fix`, `prettier --write`) on staged `*.ts` / `*.tsx` in `src/` and `web/src/`.
- **API contracts:** Zod schemas per REST endpoint → OpenAPI via `@asteasolutions/zod-to-openapi` → Scalar UI at `/api-docs`. Request bodies validated with Zod in route handlers.
- **Observability:** `@sentry/node` captures REST, WebSocket feed, and SSE errors (always on when `SENTRY_DSN` is set); `pino` JSON logs with `correlationId` (from `x-request-id` or per book tick); OpenTelemetry spans (`ws.message``orderbook.process``arbitrage.evaluate``sse.broadcast`) export to Sentry via `@sentry/opentelemetry` **only when `SENTRY_TRACING` is enabled** (default off, so 24/7 operation stays within the free-tier span quota). With tracing off the tracer is a no-op — no spans are created or exported — but errors are still captured. Dev probe: `GET /api/debug/sentry`; verify with `npm run test:observability`.
- **Rate limiting & cache:** Upstash sliding-window limits per IP on read/write/SSE routes (`429` + `Retry-After`). Latest `StateSnapshot` cached in Redis (~1s TTL) for `GET /api/state` and refreshed on SSE broadcast. Set `UPSTASH_REDIS_REST_URL` and `UPSTASH_REDIS_REST_TOKEN`. Verify with `npm run test:rate-limit`.
- **CI:** GitHub Actions quality pipeline (typecheck, unit tests, build); Playwright smoke on PRs — dashboard loads and SSE connects (`npm run test:e2e`, workflow `e2e.yml`, `DEMO_MODE=true` in CI).
- **Security scanning:** CodeQL (`.github/workflows/codeql.yml`); Dependabot for npm (root + `web/`) and GitHub Actions.
---
## Licencia
MIT — ver archivo `LICENSE` cuando se añada al repositorio.
+8
View File
@@ -0,0 +1,8 @@
import { test, expect } from "@playwright/test";
/** Asserts the dev Sentry probe responds; does not verify ingest in sentry.io (see README). */
test("sentry dev probe returns ok", async ({ request }) => {
const res = await request.get("/api/debug/sentry");
expect(res.ok()).toBeTruthy();
await expect(res.json()).resolves.toMatchObject({ ok: true });
});
+19
View File
@@ -0,0 +1,19 @@
import { test, expect } from "@playwright/test";
test("dashboard loads", async ({ page }) => {
await page.goto("/");
await expect(page.getByRole("heading", { name: /ArbPulse/i })).toBeVisible();
});
test("SSE stream connects and delivers engine state", async ({ page, request }) => {
const health = await request.get("/api/health");
expect(health.ok()).toBeTruthy();
await page.goto("/");
await expect(page.getByText("Connecting to engine…")).toBeHidden({
timeout: 45_000,
});
await expect(page.getByText("LIVE", { exact: true })).toBeVisible();
await expect(page.getByText("RUNNING")).toBeVisible();
await expect(page.getByText("Realized P&L")).toBeVisible();
});
+31
View File
@@ -0,0 +1,31 @@
import eslint from "@eslint/js";
import reactHooks from "eslint-plugin-react-hooks";
import globals from "globals";
import tseslint from "typescript-eslint";
export default tseslint.config(
{ ignores: ["web/dist/**", "data/**", "**/node_modules/**"] },
{
files: ["src/**/*.ts"],
extends: [eslint.configs.recommended, ...tseslint.configs.recommended],
languageOptions: {
ecmaVersion: 2022,
sourceType: "module",
globals: globals.node,
},
},
{
files: ["web/src/**/*.{ts,tsx}"],
extends: [eslint.configs.recommended, ...tseslint.configs.recommended],
plugins: { "react-hooks": reactHooks },
languageOptions: {
ecmaVersion: 2022,
sourceType: "module",
globals: globals.browser,
parserOptions: { ecmaFeatures: { jsx: true } },
},
rules: {
...reactHooks.configs.recommended.rules,
},
},
);
@@ -0,0 +1,2 @@
schema: spec-driven
created: 2026-07-16
@@ -0,0 +1,34 @@
## Context
`src/instrumentation/otel.ts` builds a tracer that, when `SENTRY_DSN` is set, registers a `NodeTracerProvider` with `SentrySpanProcessor`/`SentryPropagator`, so every `withSpan(...)` call on the hot path creates and exports a span to Sentry. `src/instrumentation/sentry.ts` sets `tracesSampleRate` unconditionally. Running 24/7 on a VPS multiplies span volume against Sentry's free-tier quota (5M spans/month), while error events stay small. We want tracing to remain a debugging option but be off by default.
## Goals / Non-Goals
**Goals:**
- Add a single boolean env var (`SENTRY_TRACING`, default `false`) that turns span creation/export on or off.
- Default behavior emits zero spans and zero hot-path span allocation.
- Preserve error monitoring exactly as-is regardless of the flag.
- No dependency changes; no API/wire changes.
**Non-Goals:**
- Removing OpenTelemetry or the `withSpan` abstraction.
- Making `tracesSampleRate` itself env-configurable (kept as current 0.1 prod / 1 dev when enabled).
- Changing which operations are wrapped.
## Decisions
- **Boolean gate over sample-rate gate.** A boolean `SENTRY_TRACING` that makes the tracer a no-op avoids hot-path span allocation entirely when off (aligns with the "no unnecessary alloc on the hot path" convention). A `tracesSampleRate=0` approach would still register the provider and create spans that are dropped later — more overhead. Alternative considered and rejected.
- **Gate lives in `otel.ts` init.** `initOpenTelemetry()` returns the no-op `trace.getTracer("arbpulse")` unless both `SENTRY_DSN` and `SENTRY_TRACING` are truthy. This keeps `withSpan`'s signature and all call sites unchanged; its `catch` still calls `Sentry.captureException`, so errors are captured even with a no-op span.
- **Conditional `tracesSampleRate` in `sentry.ts`.** Only set `tracesSampleRate` when tracing is enabled; omit it otherwise so Sentry performance is fully off. A shared helper `isTracingEnabled()` reads the flag so both files agree.
- **Parsing.** Reuse the existing truthy convention (`"1"` or `"true"`, case-insensitive) used by `config.ts`'s `bool()` for consistency; implement locally in the instrumentation module since it initializes before `config` import.
## Risks / Trade-offs
- [Someone enables tracing in prod and hits the quota] → Documented in `.env.example`/README that it is off by default and why; enabling is an explicit opt-in.
- [Flag read in two files could drift] → Centralize in one `isTracingEnabled()` helper exported from the instrumentation layer.
- [No-op tracer still wraps hot path in a promise] → Accepted: `withSpan` remains async as today; the change only removes span export, matching prior behavior when `SENTRY_DSN` was unset.
## Migration Plan
- Deploy with `SENTRY_TRACING` unset → tracing off (safe default), errors still reported.
- To debug, set `SENTRY_TRACING=true` (with `SENTRY_DSN`) temporarily; unset to roll back. No data migration.
@@ -0,0 +1,25 @@
## Why
Arb Pulse is moving to a 24/7 VPS (Hetzner), where the engine runs continuously and emits OpenTelemetry spans on the hot path (`ws.message``orderbook.process``arbitrage.evaluate``sse.broadcast`). At that volume the app can approach or exceed Sentry's free-tier span quota (5M spans/month), while error monitoring stays well within limits. We want to keep tracing available for debugging but off by default so continuous operation is free-tier safe.
## What Changes
- Introduce an environment variable (`SENTRY_TRACING`, boolean, default `false`) that gates OpenTelemetry span creation and export to Sentry.
- When `SENTRY_TRACING` is disabled (default), the tracer is a no-op: no spans are created or exported, and no hot-path span allocation occurs. Error capture (`Sentry.captureException`, `unhandledRejection`, `uncaughtException`) remains fully active.
- When `SENTRY_TRACING` is enabled **and** `SENTRY_DSN` is set, spans are created and exported to Sentry as today, and `tracesSampleRate` is applied.
- Document the new variable in `.env.example` and `README.md`.
## Capabilities
### New Capabilities
- `observability`: error monitoring and optional distributed tracing behavior — how Sentry error capture is always on, and how OpenTelemetry span emission is conditioned on an environment flag.
### Modified Capabilities
<!-- None: no existing specs under openspec/specs/. -->
## Impact
- Code: `src/instrumentation/otel.ts` (gate span provider/export), `src/instrumentation/sentry.ts` (conditional `tracesSampleRate`), possibly `src/instrumentation/index.ts`.
- Config/docs: `.env.example`, `README.md` (observability section + env var table).
- Dependencies: none added or removed (OTel packages stay for the enabled path).
- Behavior: default runtime emits zero spans; error monitoring unchanged. No API or wire-format changes.
@@ -0,0 +1,43 @@
## ADDED Requirements
### Requirement: Error monitoring is always active
The system SHALL report runtime errors to Sentry whenever `SENTRY_DSN` is configured, independent of any tracing configuration. This includes REST, WebSocket feed, and SSE errors captured via `Sentry.captureException`, plus process-level `unhandledRejection` and `uncaughtException` handlers.
#### Scenario: Error captured with DSN set
- **WHEN** `SENTRY_DSN` is set and a handled error occurs on the WebSocket, REST, or SSE path
- **THEN** the error is sent to Sentry as an error event
#### Scenario: No DSN configured
- **WHEN** `SENTRY_DSN` is empty or unset
- **THEN** Sentry is not initialized and no events are sent, and the application continues running normally
### Requirement: Tracing spans are gated by an environment flag
The system SHALL only create and export OpenTelemetry spans when the `SENTRY_TRACING` environment variable is enabled AND `SENTRY_DSN` is set. The flag SHALL default to disabled so that continuous 24/7 operation emits zero spans by default.
#### Scenario: Tracing disabled by default
- **WHEN** `SENTRY_TRACING` is unset or falsy (default)
- **THEN** the tracer is a no-op, no spans are created on the hot path, and no spans are exported to Sentry
#### Scenario: Tracing enabled with DSN
- **WHEN** `SENTRY_TRACING` is truthy and `SENTRY_DSN` is set
- **THEN** hot-path spans (`ws.message`, `orderbook.process`, `arbitrage.evaluate`, `sse.broadcast`) are created and exported to Sentry, and `tracesSampleRate` is applied
#### Scenario: Tracing enabled without DSN
- **WHEN** `SENTRY_TRACING` is truthy but `SENTRY_DSN` is empty or unset
- **THEN** the tracer is a no-op and no spans are exported, since export requires a DSN
### Requirement: Error capture is preserved when tracing is disabled
The system SHALL continue to report exceptions thrown inside hot-path work wrappers to Sentry even when tracing is disabled, so disabling spans never reduces error visibility.
#### Scenario: Exception in hot-path wrapper with tracing off
- **WHEN** `SENTRY_TRACING` is disabled and code wrapped by the hot-path span helper throws
- **THEN** the exception is still captured and reported to Sentry (as an error), and re-thrown to preserve existing control flow
@@ -0,0 +1,16 @@
## 1. Tracing gate
- [x] 1.1 Add `isTracingEnabled()` helper (reads `SENTRY_TRACING`, truthy = `"1"`/`"true"` case-insensitive) exported from the instrumentation layer
- [x] 1.2 In `src/instrumentation/otel.ts`, return the no-op tracer unless both `SENTRY_DSN` and `SENTRY_TRACING` are truthy (keep `withSpan` signature and its `Sentry.captureException` in catch)
- [x] 1.3 In `src/instrumentation/sentry.ts`, only set `tracesSampleRate` when tracing is enabled (omit it otherwise)
## 2. Documentation
- [x] 2.1 Add `SENTRY_TRACING` (default `false`) to `.env.example` with a note it gates spans for free-tier safety
- [x] 2.2 Update the README observability section to describe the flag and default-off behavior
## 3. Verification
- [x] 3.1 `npm run typecheck` passes
- [x] 3.2 `npm test` passes
- [x] 3.3 Manually confirm: with `SENTRY_TRACING` unset, no spans are emitted; error capture path unaffected
+20
View File
@@ -0,0 +1,20 @@
schema: spec-driven
# Project context (optional)
# This is shown to AI when creating artifacts.
# Add your tech stack, conventions, style guides, domain knowledge, etc.
# Example:
# context: |
# Tech stack: TypeScript, React, Node.js
# We use conventional commits
# Domain: e-commerce platform
# Per-artifact rules (optional)
# Add custom rules for specific artifacts.
# Example:
# rules:
# proposal:
# - Keep proposals under 500 words
# - Always include a "Non-goals" section
# tasks:
# - Break tasks into chunks of max 2 hours
+2910 -42
View File
File diff suppressed because it is too large Load Diff
+38 -6
View File
@@ -12,20 +12,52 @@
"start": "tsx src/index.ts",
"build": "npm --prefix web install && npm --prefix web run build",
"typecheck": "tsc --noEmit && npm --prefix web run typecheck",
"test": "node scripts/run-tests.mjs"
"test": "node scripts/run-tests.mjs",
"test:rate-limit": "node --import tsx --test src/interfaces/http/rate-limit.test.ts",
"test:e2e": "playwright test",
"test:observability": "playwright test observability.spec.ts",
"lint": "eslint .",
"prepare": "husky"
},
"lint-staged": {
"src/**/*.ts": [
"eslint --fix",
"prettier --write"
],
"web/src/**/*.{ts,tsx}": [
"eslint --fix",
"prettier --write"
]
},
"dependencies": {
"@asteasolutions/zod-to-openapi": "^8.5.0",
"@opentelemetry/api": "^1.9.1",
"@opentelemetry/sdk-trace-node": "^2.7.1",
"@scalar/express-api-reference": "^0.8.48",
"@sentry/node": "^10.57.0",
"@sentry/opentelemetry": "^10.57.0",
"@upstash/ratelimit": "^2.0.6",
"@upstash/redis": "^1.35.3",
"express": "^4.21.2",
"swagger-ui-express": "^5.0.1",
"pino": "^10.3.1",
"pino-pretty": "^13.1.3",
"tsx": "^4.19.2",
"ws": "^8.18.0"
"ws": "^8.18.0",
"zod": "^4.4.3"
},
"devDependencies": {
"@eslint/js": "^9.18.0",
"@playwright/test": "^1.51.1",
"@types/express": "^4.17.21",
"@types/node": "^22.10.5",
"@types/swagger-ui-express": "^4.1.8",
"@types/ws": "^8.5.13",
"openapi-types": "^12.1.3",
"typescript": "^5.7.3"
"eslint": "^9.18.0",
"eslint-plugin-react-hooks": "^5.1.0",
"globals": "^15.14.0",
"husky": "^9.1.7",
"lint-staged": "^16.4.0",
"prettier": "^3.8.3",
"typescript": "^5.7.3",
"typescript-eslint": "^8.26.0"
}
}
+24
View File
@@ -0,0 +1,24 @@
import { defineConfig, devices } from "@playwright/test";
export default defineConfig({
testDir: "e2e",
fullyParallel: false,
forbidOnly: !!process.env.CI,
retries: process.env.CI ? 2 : 0,
workers: 1,
timeout: 60_000,
reporter: process.env.CI ? "github" : "html",
use: {
baseURL: "http://localhost:8080",
trace: "on-first-retry",
},
projects: [{ name: "chromium", use: { ...devices["Desktop Chrome"] } }],
...(!process.env.CI && {
webServer: {
command: "npm start",
url: "http://localhost:8080/api/health",
reuseExistingServer: !process.env.PW_FRESH_SERVER,
env: { DEMO_MODE: "true" },
},
}),
});
@@ -1,5 +1,10 @@
import type { OrderBook } from "../../domain/entities/index.js";
import type { ArbitrageEngine } from "../../domain/services/arbitrage-engine.js";
import { withSpan } from "../../instrumentation/otel.js";
import {
newCorrelationId,
runWithCorrelation,
} from "../../infrastructure/logging/correlation.js";
export interface FeedRecorderPort {
record(book: OrderBook): void;
@@ -10,12 +15,30 @@ export class ProcessOrderBookUpdate {
constructor(
private readonly engine: ArbitrageEngine,
private readonly recorder: FeedRecorderPort,
private readonly isExchangeActive: (exchange: OrderBook["exchange"]) => boolean,
private readonly isExchangeActive: (
exchange: OrderBook["exchange"],
) => boolean,
) {}
run(book: OrderBook): void {
if (!this.isExchangeActive(book.exchange)) return;
this.recorder.record(book);
this.engine.onBook(book);
const correlationId = newCorrelationId("book");
runWithCorrelation({ correlationId, exchange: book.exchange }, () => {
void withSpan(
"orderbook.process",
{ exchange: book.exchange, correlationId },
async () => {
this.recorder.record(book);
await withSpan(
"arbitrage.evaluate",
{ exchange: book.exchange, correlationId },
async () => {
this.engine.onBook(book);
},
);
},
);
});
}
}
+19 -7
View File
@@ -1,5 +1,11 @@
import "./load-env.js";
import { initInstrumentation } from "./instrumentation/index.js";
initInstrumentation();
import express from "express";
import swaggerUi from "swagger-ui-express";
import * as Sentry from "@sentry/node";
import { apiReference } from "@scalar/express-api-reference";
import { existsSync } from "node:fs";
import { fileURLToPath } from "node:url";
import { dirname, join } from "node:path";
@@ -8,7 +14,7 @@ import { bootstrap, config } from "./composition/bootstrap.js";
import { runtime } from "./infrastructure/config/runtime.js";
import { SseHub } from "./interfaces/sse/sse.js";
import { createRouter } from "./interfaces/http/routes.js";
import { openapiSpec } from "./interfaces/http/openapi.js";
import { openapiDocument } from "./interfaces/http/openapi.js";
const log = createLogger("server");
@@ -25,11 +31,15 @@ function main(): void {
const server = express();
server.use(express.json());
server.use("/api", createRouter(ctx.application, sse));
Sentry.setupExpressErrorHandler(server);
server.use("/api-docs", swaggerUi.serve, swaggerUi.setup(openapiSpec, {
customSiteTitle: "Arb Pulse — API Docs",
swaggerOptions: { defaultModelsExpandDepth: 2, defaultModelExpandDepth: 2 },
}));
server.use(
"/api-docs",
apiReference({
content: openapiDocument,
pageTitle: "Arb Pulse — API Docs",
}),
);
if (existsSync(webDist)) {
server.use(express.static(webDist));
@@ -40,7 +50,9 @@ function main(): void {
server.get("/", (_req, res) => {
res
.status(200)
.send("Backend running. Frontend not built yet — run `npm run build` (or `npm run dev:web`).");
.send(
"Backend running. Frontend not built yet — run `npm run build` (or `npm run dev:web`).",
);
});
}
+48
View File
@@ -0,0 +1,48 @@
import { Redis } from "@upstash/redis";
const SNAPSHOT_KEY = "arbpulse:snapshot";
const SNAPSHOT_TTL_SEC = 1;
let redis: Redis | null | undefined;
export function isUpstashConfigured(): boolean {
return Boolean(
process.env.UPSTASH_REDIS_REST_URL?.trim() &&
process.env.UPSTASH_REDIS_REST_TOKEN?.trim(),
);
}
function getRedis(): Redis | null {
if (redis !== undefined) {
return redis;
}
const url = process.env.UPSTASH_REDIS_REST_URL?.trim();
const token = process.env.UPSTASH_REDIS_REST_TOKEN?.trim();
if (!url || !token) {
redis = null;
return null;
}
redis = new Redis({ url, token });
return redis;
}
export async function getCachedSnapshot<T>(): Promise<T | null> {
const client = getRedis();
if (!client) return null;
try {
return (await client.get<T>(SNAPSHOT_KEY)) ?? null;
} catch (err) {
console.warn("[upstash] get snapshot failed", err);
return null;
}
}
export async function setCachedSnapshot<T>(snapshot: T): Promise<void> {
const client = getRedis();
if (!client) return;
try {
await client.set(SNAPSHOT_KEY, snapshot, { ex: SNAPSHOT_TTL_SEC });
} catch (err) {
console.warn("[upstash] set snapshot failed", err);
}
}
+33 -10
View File
@@ -1,5 +1,7 @@
import WebSocket from "ws";
import * as Sentry from "@sentry/node";
import type { ExchangeId, OrderBook } from "../../domain/entities/index.js";
import { withSpan } from "../../instrumentation/otel.js";
import { createLogger, type Logger } from "../logging/logger.js";
import type { MarketDataFeed } from "../../domain/ports/ports.js";
import { LocalBook } from "./local-book.js";
@@ -56,12 +58,16 @@ export abstract class ExchangeConnector implements MarketDataFeed {
ws.on("open", () => {
this.reconnectAttempts = 0;
this.book.reset();
this.log.info(this.skipSubscribe() ? "connected" : "connected, subscribing");
this.log.info(
this.skipSubscribe() ? "connected" : "connected, subscribing",
);
if (!this.skipSubscribe()) {
try {
ws.send(JSON.stringify(this.subscribeMessage()));
} catch (err) {
this.log.error("subscribe send failed", err);
this.log.error("subscribe send failed", {
error: err instanceof Error ? err.message : String(err),
});
}
}
this.startPing();
@@ -72,15 +78,27 @@ export abstract class ExchangeConnector implements MarketDataFeed {
const text = data.toString();
// Some exchanges (OKX) reply to app-level pings with a plain "pong" frame.
if (text === "pong" || text === "ping") return;
try {
this.handleMessage(JSON.parse(text));
} catch (err) {
this.log.warn("failed to parse message", err);
}
void withSpan(
"ws.message",
{ exchange: this.id, bytes: text.length },
async () => {
try {
this.handleMessage(JSON.parse(text));
} catch (err) {
this.log.warn("failed to parse message", {
error: err instanceof Error ? err.message : String(err),
});
}
},
);
});
ws.on("error", (err) => {
this.log.error("socket error", err instanceof Error ? err.message : err);
const error = err instanceof Error ? err : new Error(String(err));
this.log.error("socket error", { error: error.message });
Sentry.captureException(error, {
tags: { exchange: this.id, component: "ws" },
});
});
ws.on("close", () => {
@@ -97,8 +115,13 @@ export abstract class ExchangeConnector implements MarketDataFeed {
private scheduleReconnect(): void {
this.reconnectAttempts += 1;
const delay = Math.min(30_000, 500 * 2 ** Math.min(this.reconnectAttempts, 6));
this.log.warn(`disconnected, reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
const delay = Math.min(
30_000,
500 * 2 ** Math.min(this.reconnectAttempts, 6),
);
this.log.warn(
`disconnected, reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`,
);
setTimeout(() => {
if (!this.closed) this.connect();
}, delay);
+39
View File
@@ -0,0 +1,39 @@
import { AsyncLocalStorage } from "node:async_hooks";
import { randomUUID } from "node:crypto";
export interface CorrelationContext {
correlationId: string;
exchange?: string;
}
const storage = new AsyncLocalStorage<CorrelationContext>();
export function getCorrelationContext(): CorrelationContext | undefined {
return storage.getStore();
}
export function getCorrelationId(): string | undefined {
return storage.getStore()?.correlationId;
}
export function runWithCorrelation<T>(
context: CorrelationContext,
fn: () => T,
): T {
return storage.run(context, fn);
}
export function newCorrelationId(prefix = "tick"): string {
return `${prefix}_${randomUUID().slice(0, 8)}`;
}
export function bindRequestCorrelation(
headerValue: string | string[] | undefined,
): CorrelationContext {
const raw = Array.isArray(headerValue) ? headerValue[0] : headerValue;
const correlationId =
typeof raw === "string" && raw.trim()
? raw.trim()
: newCorrelationId("req");
return { correlationId };
}
+31 -15
View File
@@ -1,22 +1,38 @@
type Level = "info" | "warn" | "error";
import pino from "pino";
import { getCorrelationContext } from "./correlation.js";
function emit(level: Level, scope: string, msg: string, extra?: unknown): void {
const ts = new Date().toISOString();
const line = `${ts} [${level.toUpperCase()}] (${scope}) ${msg}`;
if (level === "error") {
console.error(line, extra ?? "");
} else if (level === "warn") {
console.warn(line, extra ?? "");
} else {
console.log(line, extra ?? "");
}
}
const root = pino({
level: process.env.LOG_LEVEL ?? "info",
base: { service: "arbpulse" },
timestamp: pino.stdTimeFunctions.isoTime,
...(process.env.NODE_ENV !== "production"
? {
transport: {
target: "pino-pretty",
options: { colorize: true, singleLine: true },
},
}
: {}),
});
export function createLogger(scope: string) {
const child = root.child({ scope });
return {
info: (msg: string, extra?: unknown) => emit("info", scope, msg, extra),
warn: (msg: string, extra?: unknown) => emit("warn", scope, msg, extra),
error: (msg: string, extra?: unknown) => emit("error", scope, msg, extra),
info: (msg: string, extra?: Record<string, unknown>) =>
child.info({ ...bindings(), ...extra }, msg),
warn: (msg: string, extra?: Record<string, unknown>) =>
child.warn({ ...bindings(), ...extra }, msg),
error: (msg: string, extra?: Record<string, unknown>) =>
child.error({ ...bindings(), ...extra }, msg),
};
}
function bindings(): Record<string, unknown> {
const ctx = getCorrelationContext();
if (!ctx) return {};
return {
correlationId: ctx.correlationId,
...(ctx.exchange ? { exchange: ctx.exchange } : {}),
};
}
+7
View File
@@ -0,0 +1,7 @@
import { initOpenTelemetry } from "./otel.js";
import { initSentry } from "./sentry.js";
export function initInstrumentation(): void {
initSentry();
initOpenTelemetry();
}
+55
View File
@@ -0,0 +1,55 @@
import * as Sentry from "@sentry/node";
import { trace, type Span, type Tracer } from "@opentelemetry/api";
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { SentrySpanProcessor, SentryPropagator } from "@sentry/opentelemetry";
import { isTracingEnabled } from "./tracing.js";
let tracer: Tracer | null = null;
/**
* OpenTelemetry spans export to Sentry via @sentry/opentelemetry only when
* SENTRY_DSN is set AND SENTRY_TRACING is enabled. Otherwise spans are no-ops
* (default; safe for 24/7 operation, local dev, and tests). Error capture in
* `withSpan` is preserved regardless.
*/
export function initOpenTelemetry(): Tracer {
if (tracer) return tracer;
if (!process.env.SENTRY_DSN?.trim() || !isTracingEnabled()) {
tracer = trace.getTracer("arbpulse");
return tracer;
}
const provider = new NodeTracerProvider({
spanProcessors: [new SentrySpanProcessor()],
});
provider.register({
propagator: new SentryPropagator(),
});
tracer = provider.getTracer("arbpulse");
return tracer;
}
export function getTracer(): Tracer {
return tracer ?? initOpenTelemetry();
}
export async function withSpan<T>(
name: string,
attributes: Record<string, string | number | boolean>,
fn: (span: Span) => T | Promise<T>,
): Promise<T> {
const activeTracer = getTracer();
return activeTracer.startActiveSpan(name, { attributes }, async (span) => {
try {
return await fn(span);
} catch (err) {
span.setStatus({ code: 2, message: String(err) });
Sentry.captureException(err);
throw err;
} finally {
span.end();
}
});
}
+32
View File
@@ -0,0 +1,32 @@
import * as Sentry from "@sentry/node";
import { isTracingEnabled } from "./tracing.js";
const dsn = process.env.SENTRY_DSN?.trim();
export function initSentry(): void {
if (!dsn) return;
Sentry.init({
dsn,
environment: process.env.NODE_ENV ?? "development",
integrations: [Sentry.httpIntegration(), Sentry.expressIntegration()],
// Tracing/spans are gated by SENTRY_TRACING (default off) to stay within
// the Sentry free-tier span quota during 24/7 operation. Error monitoring
// is always on.
...(isTracingEnabled()
? { tracesSampleRate: process.env.NODE_ENV === "production" ? 0.1 : 1 }
: {}),
});
process.on("unhandledRejection", (reason) => {
Sentry.captureException(
reason instanceof Error ? reason : new Error(String(reason)),
);
});
process.on("uncaughtException", (error) => {
Sentry.captureException(error);
});
}
export { Sentry };
+11
View File
@@ -0,0 +1,11 @@
/**
* Single source of truth for the tracing gate. OpenTelemetry spans are only
* created and exported to Sentry when `SENTRY_TRACING` is truthy (and a
* `SENTRY_DSN` is set). Defaults to disabled so 24/7 operation stays off the
* Sentry free-tier span quota. Error monitoring is unaffected by this flag.
*/
export function isTracingEnabled(): boolean {
const raw = process.env.SENTRY_TRACING;
if (raw === undefined) return false;
return raw === "1" || raw.toLowerCase() === "true";
}
File diff suppressed because it is too large Load Diff
+60
View File
@@ -0,0 +1,60 @@
import assert from "node:assert/strict";
import { randomUUID } from "node:crypto";
import type { AddressInfo } from "node:net";
import { test } from "node:test";
import express, { Router } from "express";
import { isUpstashConfigured } from "../../infrastructure/cache/upstash.js";
import { loadEnvLocal } from "../../test-support/load-env-local.js";
import { createRateLimitMiddleware } from "./rate-limit.js";
loadEnvLocal();
const READ_LIMIT_PER_MIN = 60;
test(
"read tier returns 429 with Retry-After after limit",
{ skip: !isUpstashConfigured() },
async () => {
const app = express();
const router = Router();
router.use(createRateLimitMiddleware());
router.get("/state", (_req, res) => {
res.json({ success: true, data: { ok: true } });
});
app.use("/api", router);
const server = await new Promise<ReturnType<typeof app.listen>>(
(resolve) => {
const s = app.listen(0, "127.0.0.1", () => resolve(s));
},
);
const port = (server.address() as AddressInfo).port;
const url = `http://127.0.0.1:${port}/api/state`;
const clientIp = `rate-limit-qa-${randomUUID()}`;
try {
for (let i = 0; i < READ_LIMIT_PER_MIN; i++) {
const res = await fetch(url, {
headers: { "x-forwarded-for": clientIp },
});
assert.equal(res.status, 200, `request ${i + 1} should pass`);
}
const blocked = await fetch(url, {
headers: { "x-forwarded-for": clientIp },
});
assert.equal(blocked.status, 429);
assert.ok(blocked.headers.get("retry-after"));
const body = (await blocked.json()) as { error?: string };
assert.equal(body.error, "Too many requests");
} finally {
await new Promise<void>((resolve, reject) => {
server.close((err) => (err ? reject(err) : resolve()));
});
}
},
);
test("reports whether Upstash REST credentials are set", () => {
assert.equal(typeof isUpstashConfigured(), "boolean");
});
+86
View File
@@ -0,0 +1,86 @@
import { Ratelimit } from "@upstash/ratelimit";
import { Redis } from "@upstash/redis";
import type { NextFunction, Request, Response } from "express";
import { isUpstashConfigured } from "../../infrastructure/cache/upstash.js";
type RouteTier = "read" | "stream" | "write" | "health";
const LIMITS: Record<
Exclude<RouteTier, "health">,
{ requests: number; window: `${number} s` | `${number} m` }
> = {
read: { requests: 60, window: "1 m" },
stream: { requests: 10, window: "1 m" },
write: { requests: 30, window: "1 m" },
};
const limiters = new Map<Exclude<RouteTier, "health">, Ratelimit>();
function getLimiter(tier: Exclude<RouteTier, "health">): Ratelimit | null {
if (!isUpstashConfigured()) {
return null;
}
let limiter = limiters.get(tier);
if (limiter) {
return limiter;
}
const url = process.env.UPSTASH_REDIS_REST_URL!.trim();
const token = process.env.UPSTASH_REDIS_REST_TOKEN!.trim();
const redis = new Redis({ url, token });
const cfg = LIMITS[tier];
limiter = new Ratelimit({
redis,
limiter: Ratelimit.slidingWindow(cfg.requests, cfg.window),
prefix: `arbpulse:${tier}`,
});
limiters.set(tier, limiter);
return limiter;
}
function clientKey(req: Request): string {
const forwarded = req.header("x-forwarded-for");
if (forwarded) {
return forwarded.split(",")[0]?.trim() || "unknown";
}
return req.ip || req.socket.remoteAddress || "unknown";
}
function tierForPath(path: string, method: string): RouteTier {
if (path === "/health") return "health";
if (path === "/stream") return "stream";
if (method === "GET" || method === "HEAD") return "read";
return "write";
}
export function createRateLimitMiddleware() {
return async (
req: Request,
res: Response,
next: NextFunction,
): Promise<void> => {
const tier = tierForPath(req.path, req.method);
if (tier === "health") {
next();
return;
}
const limiter = getLimiter(tier);
if (!limiter) {
next();
return;
}
const { success, reset } = await limiter.limit(`${tier}:${clientKey(req)}`);
if (success) {
next();
return;
}
const retryAfterSec = Math.max(1, Math.ceil((reset - Date.now()) / 1000));
res.setHeader("Retry-After", String(retryAfterSec));
res.status(429).json({
success: false,
error: "Too many requests",
});
};
}
+75 -32
View File
@@ -1,16 +1,64 @@
import { Router, type Request, type Response } from "express";
import {
Router,
type NextFunction,
type Request,
type Response,
} from "express";
import { Sentry } from "../../instrumentation/sentry.js";
import type { ApplicationService } from "../../composition/application-service.js";
import {
bindRequestCorrelation,
runWithCorrelation,
} from "../../infrastructure/logging/correlation.js";
import {
getCachedSnapshot,
setCachedSnapshot,
} from "../../infrastructure/cache/upstash.js";
import type { SseHub } from "../sse/sse.js";
import { createRateLimitMiddleware } from "./rate-limit.js";
import {
ConfigPatchSchema,
DemoControlBodySchema,
MaxTradeControlBodySchema,
RecordControlBodySchema,
ThresholdControlBodySchema,
} from "./schemas/requests.js";
import { parseBody } from "./validate.js";
export function createRouter(app: ApplicationService, sse: SseHub): Router {
const router = Router();
router.use((req: Request, res: Response, next: NextFunction) => {
const ctx = bindRequestCorrelation(req.header("x-request-id"));
res.setHeader("x-request-id", ctx.correlationId);
runWithCorrelation(ctx, () => next());
});
router.use(createRateLimitMiddleware());
router.get("/health", (_req: Request, res: Response) => {
res.json({ success: true, data: { status: "ok", ts: Date.now() } });
});
router.get("/state", (_req: Request, res: Response) => {
res.json({ success: true, data: app.getSnapshot() });
router.get("/debug/sentry", (_req: Request, res: Response) => {
if (process.env.NODE_ENV === "production") {
res.status(404).json({ error: "Not found" });
return;
}
Sentry.captureException(new Error("Arb Pulse Sentry test error"));
res.json({ ok: true, message: "Test error sent to Sentry" });
});
router.get("/state", async (_req: Request, res: Response) => {
const cached =
await getCachedSnapshot<ReturnType<ApplicationService["getSnapshot"]>>();
if (cached) {
res.json({ success: true, data: cached });
return;
}
const snapshot = app.getSnapshot();
void setCachedSnapshot(snapshot);
res.json({ success: true, data: snapshot });
});
router.get("/stream", (req: Request, res: Response) => {
@@ -22,7 +70,10 @@ export function createRouter(app: ApplicationService, sse: SseHub): Router {
});
router.patch("/config", (req: Request, res: Response) => {
const error = app.patchConfig(req.body ?? {});
const patch = parseBody(ConfigPatchSchema, req, res);
if (patch === null) return;
const error = app.patchConfig(patch);
if (error) {
res.status(400).json({ success: false, error });
return;
@@ -46,51 +97,43 @@ export function createRouter(app: ApplicationService, sse: SseHub): Router {
});
router.post("/control/demo", (req: Request, res: Response) => {
const enabled = req.body?.enabled;
if (typeof enabled !== "boolean") {
res.status(400).json({ success: false, error: "enabled must be a boolean" });
return;
}
app.setDemoMode(enabled);
res.json({ success: true, data: { demoMode: enabled } });
const body = parseBody(DemoControlBodySchema, req, res);
if (body === null) return;
app.setDemoMode(body.enabled);
res.json({ success: true, data: { demoMode: body.enabled } });
});
router.post("/control/record", (req: Request, res: Response) => {
const enabled = req.body?.enabled;
if (typeof enabled !== "boolean") {
res.status(400).json({ success: false, error: "enabled must be a boolean" });
return;
}
app.setRecordFeed(enabled);
res.json({ success: true, data: { recordFeed: enabled } });
const body = parseBody(RecordControlBodySchema, req, res);
if (body === null) return;
app.setRecordFeed(body.enabled);
res.json({ success: true, data: { recordFeed: body.enabled } });
});
router.post("/control/threshold", (req: Request, res: Response) => {
const pct = req.body?.pct;
if (typeof pct !== "number" || !Number.isFinite(pct)) {
res.status(400).json({ success: false, error: "pct must be a finite number" });
return;
}
const error = app.setThreshold(pct);
const body = parseBody(ThresholdControlBodySchema, req, res);
if (body === null) return;
const error = app.setThreshold(body.pct);
if (error) {
res.status(400).json({ success: false, error });
return;
}
res.json({ success: true, data: { minNetProfitPct: pct } });
res.json({ success: true, data: { minNetProfitPct: body.pct } });
});
router.post("/control/max-trade", (req: Request, res: Response) => {
const btc = req.body?.btc;
if (typeof btc !== "number" || !Number.isFinite(btc)) {
res.status(400).json({ success: false, error: "btc must be a finite number" });
return;
}
const error = app.setMaxTradeBtc(btc);
const body = parseBody(MaxTradeControlBodySchema, req, res);
if (body === null) return;
const error = app.setMaxTradeBtc(body.btc);
if (error) {
res.status(400).json({ success: false, error });
return;
}
res.json({ success: true, data: { maxTradeBtc: btc } });
res.json({ success: true, data: { maxTradeBtc: body.btc } });
});
return router;
+59
View File
@@ -0,0 +1,59 @@
import { extendZodWithOpenApi } from "@asteasolutions/zod-to-openapi";
import { z } from "zod";
extendZodWithOpenApi(z);
export { z };
export const ExchangeIdSchema = z
.enum(["kraken", "bybit", "okx", "binance"])
.openapi("ExchangeId");
export const FeedStatusSchema = z
.enum(["connecting", "live", "stale", "down"])
.openapi("FeedStatus");
export const CircuitStateSchema = z
.enum(["running", "paused", "tripped"])
.openapi("CircuitState");
export const OpportunityStatusSchema = z
.enum([
"executed",
"executed_partial",
"rejected_fees",
"rejected_liquidity",
"rejected_risk",
"rejected_flicker",
"rejected_stale",
"pending_confirm",
])
.openapi("OpportunityStatus");
export const ErrorEnvelopeSchema = z
.object({
success: z.literal(false),
error: z.string(),
})
.strict()
.openapi("ErrorEnvelope");
export const SuccessEnvelopeNoDataSchema = z
.object({
success: z.literal(true),
})
.strict()
.openapi("SuccessEnvelopeNoData");
export function successEnvelope<T extends z.ZodType>(
dataSchema: T,
name: string,
) {
return z
.object({
success: z.literal(true),
data: dataSchema,
})
.strict()
.openapi(name);
}
+161
View File
@@ -0,0 +1,161 @@
import {
CircuitStateSchema,
ExchangeIdSchema,
FeedStatusSchema,
OpportunityStatusSchema,
z,
} from "./common.js";
export const BestQuoteSchema = z
.object({
exchange: ExchangeIdSchema,
bid: z.number().nullable(),
bidQty: z.number().nullable(),
ask: z.number().nullable(),
askQty: z.number().nullable(),
recvTs: z.number().int().nullable(),
status: FeedStatusSchema,
ageMs: z.number().int().nullable(),
})
.strict()
.openapi("BestQuote");
export const WalletSchema = z
.object({
exchange: ExchangeIdSchema,
usdt: z.number(),
btc: z.number(),
})
.strict()
.openapi("Wallet");
export const OpportunitySchema = z
.object({
id: z.string(),
ts: z.number().int(),
buyExchange: ExchangeIdSchema,
sellExchange: ExchangeIdSchema,
topBuyAsk: z.number(),
topSellBid: z.number(),
volumeBtc: z.number(),
buyVwap: z.number(),
sellVwap: z.number(),
grossSpread: z.number(),
grossSpreadPct: z.number(),
feeBuy: z.number(),
feeSell: z.number(),
netProfit: z.number(),
netProfitPct: z.number(),
status: OpportunityStatusSchema,
reason: z.string(),
demo: z.boolean(),
})
.strict()
.openapi("Opportunity");
export const TradeSchema = z
.object({
id: z.string(),
ts: z.number().int(),
buyExchange: ExchangeIdSchema,
sellExchange: ExchangeIdSchema,
volumeBtc: z.number(),
requestedBtc: z.number(),
buyVwap: z.number(),
sellVwap: z.number(),
execBuyVwap: z.number(),
execSellVwap: z.number(),
feeBuy: z.number(),
feeSell: z.number(),
netProfit: z.number(),
netProfitPct: z.number(),
partial: z.boolean(),
demo: z.boolean(),
})
.strict()
.openapi("Trade");
export const RebalanceEventSchema = z
.object({
id: z.string(),
ts: z.number().int(),
fromExchange: ExchangeIdSchema,
toExchange: ExchangeIdSchema,
asset: z.enum(["BTC", "USDT"]),
amount: z.number(),
withdrawalFee: z.number(),
reason: z.string(),
})
.strict()
.openapi("RebalanceEvent");
export const PnlPointSchema = z
.object({
ts: z.number().int(),
pnl: z.number(),
})
.strict()
.openapi("PnlPoint");
export const EngineStatsSchema = z
.object({
uptimeMs: z.number().int(),
ticksProcessed: z.number().int(),
opportunitiesDetected: z.number().int(),
tradesExecuted: z.number().int(),
tradesRejected: z.number().int(),
realizedPnl: z.number(),
consecutiveLosses: z.number().int(),
circuit: CircuitStateSchema,
demoMode: z.boolean(),
avgTickMs: z.number(),
})
.strict()
.openapi("EngineStats");
const ActiveExchangesSchema = z.record(ExchangeIdSchema, z.boolean());
export const PublicConfigSchema = z
.object({
minNetProfitPct: z.number(),
maxTradeBtc: z.number(),
staleMs: z.number().int(),
flickerConfirmMs: z.number().int(),
latencyMs: z.number().int(),
activeExchanges: ActiveExchangesSchema,
defaults: z
.object({
minNetProfitPct: z.number(),
maxTradeBtc: z.number(),
flickerConfirmMs: z.number().int(),
activeExchanges: ActiveExchangesSchema,
})
.strict(),
takerFees: z.record(ExchangeIdSchema, z.number()),
withdrawalFeesBtc: z.record(ExchangeIdSchema, z.number()),
})
.strict()
.openapi("PublicConfig");
export const StateSnapshotSchema = z
.object({
ts: z.number().int(),
quotes: z.array(BestQuoteSchema),
wallets: z.array(WalletSchema),
stats: EngineStatsSchema,
recentOpportunities: z.array(OpportunitySchema),
recentTrades: z.array(TradeSchema),
rebalances: z.array(RebalanceEventSchema),
pnlSeries: z.array(PnlPointSchema),
config: PublicConfigSchema,
})
.strict()
.openapi("StateSnapshot");
export const HealthDataSchema = z
.object({
status: z.literal("ok"),
ts: z.number().int(),
})
.strict()
.openapi("HealthData");
+53
View File
@@ -0,0 +1,53 @@
import { ExchangeIdSchema, z } from "./common.js";
const MIN_PROFIT_PCT = 0.0001;
const MAX_PROFIT_PCT = 0.01;
const MIN_TRADE_BTC = 0.01;
const MAX_TRADE_BTC = 1.0;
const MAX_FLICKER_MS = 500;
export const BooleanControlBodySchema = z
.object({
enabled: z.boolean(),
})
.strict()
.openapi("BooleanControlBody");
export const DemoControlBodySchema =
BooleanControlBodySchema.openapi("DemoControlBody");
export const RecordControlBodySchema =
BooleanControlBodySchema.openapi("RecordControlBody");
export const ThresholdControlBodySchema = z
.object({
pct: z.number().finite().min(MIN_PROFIT_PCT).max(MAX_PROFIT_PCT),
})
.strict()
.openapi("ThresholdControlBody");
export const MaxTradeControlBodySchema = z
.object({
btc: z.number().finite().min(MIN_TRADE_BTC).max(MAX_TRADE_BTC),
})
.strict()
.openapi("MaxTradeControlBody");
export const ConfigPatchSchema = z
.object({
minNetProfitPct: z
.number()
.finite()
.min(MIN_PROFIT_PCT)
.max(MAX_PROFIT_PCT)
.optional(),
maxTradeBtc: z
.number()
.finite()
.min(MIN_TRADE_BTC)
.max(MAX_TRADE_BTC)
.optional(),
flickerConfirmMs: z.number().int().min(0).max(MAX_FLICKER_MS).optional(),
activeExchanges: z.record(ExchangeIdSchema, z.boolean()).optional(),
})
.strict()
.openapi("ConfigPatch");
+42
View File
@@ -0,0 +1,42 @@
import { successEnvelope, SuccessEnvelopeNoDataSchema } from "./common.js";
import {
HealthDataSchema,
PublicConfigSchema,
StateSnapshotSchema,
} from "./domain.js";
import { z } from "./common.js";
export const HealthResponseSchema = successEnvelope(
HealthDataSchema,
"HealthResponse",
);
export const StateResponseSchema = successEnvelope(
StateSnapshotSchema,
"StateResponse",
);
export const ConfigResponseSchema = successEnvelope(
PublicConfigSchema,
"ConfigResponse",
);
export const DemoModeResponseSchema = successEnvelope(
z.object({ demoMode: z.boolean() }).strict(),
"DemoModeResponse",
);
export const RecordFeedResponseSchema = successEnvelope(
z.object({ recordFeed: z.boolean() }).strict(),
"RecordFeedResponse",
);
export const ThresholdResponseSchema = successEnvelope(
z.object({ minNetProfitPct: z.number() }).strict(),
"ThresholdResponse",
);
export const MaxTradeResponseSchema = successEnvelope(
z.object({ maxTradeBtc: z.number() }).strict(),
"MaxTradeResponse",
);
export { SuccessEnvelopeNoDataSchema };
+18
View File
@@ -0,0 +1,18 @@
import type { Request, Response } from "express";
import type { ZodType } from "zod";
export function parseBody<T>(
schema: ZodType<T>,
req: Request,
res: Response,
): T | null {
const result = schema.safeParse(req.body ?? {});
if (!result.success) {
const message = result.error.issues
.map((issue) => issue.message)
.join("; ");
res.status(400).json({ success: false, error: message });
return null;
}
return result.data;
}
+50 -15
View File
@@ -1,6 +1,9 @@
import type { Request, Response } from "express";
import * as Sentry from "@sentry/node";
import { config } from "../../infrastructure/config/config.js";
import { setCachedSnapshot } from "../../infrastructure/cache/upstash.js";
import { createLogger } from "../../infrastructure/logging/logger.js";
import { withSpan } from "../../instrumentation/otel.js";
import type { ApplicationService } from "../../composition/application-service.js";
const log = createLogger("sse");
@@ -22,26 +25,58 @@ export class SseHub {
}
handle(req: Request, res: Response): void {
res.writeHead(200, {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache, no-transform",
Connection: "keep-alive",
"X-Accel-Buffering": "no",
});
res.write(`retry: 2000\n\n`);
this.send(res, this.app.getSnapshot());
this.clients.add(res);
log.info(`client connected (${this.clients.size} total)`);
try {
res.writeHead(200, {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache, no-transform",
Connection: "keep-alive",
"X-Accel-Buffering": "no",
});
res.write(`retry: 2000\n\n`);
this.send(res, this.app.getSnapshot());
this.clients.add(res);
log.info(`client connected`, { clientCount: this.clients.size });
req.on("close", () => {
this.clients.delete(res);
});
req.on("close", () => {
this.clients.delete(res);
});
res.on("error", (err) => {
const error = err instanceof Error ? err : new Error(String(err));
log.error("sse client error", { error: error.message });
Sentry.captureException(error, { tags: { component: "sse" } });
this.clients.delete(res);
});
} catch (err) {
const error = err instanceof Error ? err : new Error(String(err));
log.error("sse handle failed", { error: error.message });
Sentry.captureException(error, { tags: { component: "sse" } });
if (!res.headersSent) {
res.status(500).json({ success: false, error: "SSE setup failed" });
}
}
}
private broadcast(): void {
if (this.clients.size === 0) return;
const snapshot = this.app.getSnapshot();
for (const res of this.clients) this.send(res, snapshot);
void withSpan(
"sse.broadcast",
{ clientCount: this.clients.size },
async () => {
const snapshot = this.app.getSnapshot();
void setCachedSnapshot(snapshot);
for (const res of this.clients) {
try {
this.send(res, snapshot);
} catch (err) {
const error = err instanceof Error ? err : new Error(String(err));
log.warn("sse send failed", { error: error.message });
Sentry.captureException(error, { tags: { component: "sse" } });
this.clients.delete(res);
}
}
},
);
}
private send(res: Response, data: unknown): void {
+9
View File
@@ -0,0 +1,9 @@
import { existsSync } from "node:fs";
import { loadEnvFile } from "node:process";
for (const file of [".env.local", ".env"]) {
if (existsSync(file)) {
loadEnvFile(file);
break;
}
}
+35
View File
@@ -0,0 +1,35 @@
import { existsSync, readFileSync } from "node:fs";
import { join } from "node:path";
/** Loads `.env.local` without overriding variables already set in the shell. */
export function loadEnvLocal(): void {
const file = join(process.cwd(), ".env.local");
if (!existsSync(file)) {
return;
}
for (const line of readFileSync(file, "utf8").split(/\r?\n/)) {
const trimmed = line.trim();
if (!trimmed || trimmed.startsWith("#")) {
continue;
}
const eq = trimmed.indexOf("=");
if (eq <= 0) {
continue;
}
const key = trimmed.slice(0, eq).trim();
let value = trimmed.slice(eq + 1).trim();
if (
(value.startsWith('"') && value.endsWith('"')) ||
(value.startsWith("'") && value.endsWith("'"))
) {
value = value.slice(1, -1);
}
if (process.env[key] === undefined) {
process.env[key] = value;
}
}
}
+30 -24
View File
@@ -9,11 +9,11 @@
"version": "1.0.0",
"dependencies": {
"arb-pulse": "file:..",
"react": "^18.3.1",
"react": "^19.2.7",
"react-dom": "^18.3.1"
},
"devDependencies": {
"@types/react": "^18.3.18",
"@types/react": "^19.2.17",
"@types/react-dom": "^18.3.5",
"@vitejs/plugin-react": "^4.3.4",
"autoprefixer": "^10.4.20",
@@ -27,18 +27,35 @@
"name": "arb-pulse",
"version": "1.0.0",
"dependencies": {
"@asteasolutions/zod-to-openapi": "^8.5.0",
"@opentelemetry/api": "^1.9.1",
"@opentelemetry/sdk-trace-node": "^2.7.1",
"@scalar/express-api-reference": "^0.8.48",
"@sentry/node": "^10.57.0",
"@sentry/opentelemetry": "^10.57.0",
"@upstash/ratelimit": "^2.0.6",
"@upstash/redis": "^1.35.3",
"express": "^4.21.2",
"swagger-ui-express": "^5.0.1",
"pino": "^10.3.1",
"pino-pretty": "^13.1.3",
"tsx": "^4.19.2",
"ws": "^8.18.0"
"ws": "^8.18.0",
"zod": "^4.4.3"
},
"devDependencies": {
"@eslint/js": "^9.18.0",
"@playwright/test": "^1.51.1",
"@types/express": "^4.17.21",
"@types/node": "^22.10.5",
"@types/swagger-ui-express": "^4.1.8",
"@types/ws": "^8.5.13",
"openapi-types": "^12.1.3",
"typescript": "^5.7.3"
"eslint": "^9.18.0",
"eslint-plugin-react-hooks": "^5.1.0",
"globals": "^15.14.0",
"husky": "^9.1.7",
"lint-staged": "^16.4.0",
"prettier": "^3.8.3",
"typescript": "^5.7.3",
"typescript-eslint": "^8.26.0"
},
"engines": {
"node": ">=20"
@@ -1278,21 +1295,13 @@
"dev": true,
"license": "MIT"
},
"node_modules/@types/prop-types": {
"version": "15.7.15",
"resolved": "https://registry.npmjs.org/@types/prop-types/-/prop-types-15.7.15.tgz",
"integrity": "sha512-F6bEyamV9jKGAFBEmlQnesRPGOQqS2+Uwi0Em15xenOxHaf2hv6L8YCVn3rPdPJOiJfPiCnLIRyvwVaqMY3MIw==",
"dev": true,
"license": "MIT"
},
"node_modules/@types/react": {
"version": "18.3.31",
"resolved": "https://registry.npmjs.org/@types/react/-/react-18.3.31.tgz",
"integrity": "sha512-vfEqpXTvwT91yhmwdfouStN2hSKwTvyRs8qpLfADyrq/kxDw0hZM7Wk9Ug1FELj8hIby+S/+kQCSRFF32nv2Qw==",
"version": "19.2.17",
"resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.17.tgz",
"integrity": "sha512-MXfmqaVPEVgkBT/aY0aGCkRWWtByiYQXo3xdQ8r5RzuFrPiRn8Gar2tQdXSUQ2GKV3bkXckek89V8wQBY2Q/Aw==",
"dev": true,
"license": "MIT",
"dependencies": {
"@types/prop-types": "*",
"csstype": "^3.2.2"
}
},
@@ -2284,13 +2293,10 @@
"license": "MIT"
},
"node_modules/react": {
"version": "18.3.1",
"resolved": "https://registry.npmjs.org/react/-/react-18.3.1.tgz",
"integrity": "sha512-wS+hAgJShR0KhEvPJArfuPVN1+Hz1t0Y6n5jLrGQbkb4urgPE/0Rve+1kMB1v/oWgHgm4WIcV+i7F2pTVj+2iQ==",
"version": "19.2.7",
"resolved": "https://registry.npmjs.org/react/-/react-19.2.7.tgz",
"integrity": "sha512-HNe9WslTbXmFK8o8cmwgAeJFSBvt1bPdHCVKtaaV+WlAN36mpT4hcRpwbf3fY56ar2oIXzsBpOAiIRHAdY0OlQ==",
"license": "MIT",
"dependencies": {
"loose-envify": "^1.1.0"
},
"engines": {
"node": ">=0.10.0"
}
+2 -2
View File
@@ -11,11 +11,11 @@
},
"dependencies": {
"arb-pulse": "file:..",
"react": "^18.3.1",
"react": "^19.2.7",
"react-dom": "^18.3.1"
},
"devDependencies": {
"@types/react": "^18.3.18",
"@types/react": "^19.2.17",
"@types/react-dom": "^18.3.5",
"@vitejs/plugin-react": "^4.3.4",
"autoprefixer": "^10.4.20",