From b8639f0dfdefd37089f5c62c98262873914b538e Mon Sep 17 00:00:00 2001 From: Vortex Backend Date: Sun, 30 Aug 2026 05:27:50 +0100 Subject: [PATCH] feat: add ws backplane, health probes, and treasury reporting --- .env.example | 6 + CONTRIBUTING.md | 8 +- README.md | 16 ++ docker-compose.yml | 42 +++++ prisma/schema.prisma | 2 + src/config/configuration.ts | 4 + src/config/env.validation.ts | 3 + src/health/health.controller.ts | 25 +++ src/intents/intents.controller.ts | 3 + src/intents/intents.gateway.ts | 185 +++++++++++++++++++++-- src/intents/intents.types.ts | 1 + src/intents/prisma-intents.repository.ts | 13 +- src/stats/stats.controller.ts | 5 + src/stats/stats.service.ts | 62 ++++++++ 14 files changed, 363 insertions(+), 12 deletions(-) create mode 100644 docker-compose.yml diff --git a/.env.example b/.env.example index 489357c..52dbcb2 100644 --- a/.env.example +++ b/.env.example @@ -55,3 +55,9 @@ CORS_ORIGIN=* # Connections beyond this limit are rejected with close code 1013 (try again later). # Set to 0 to disable the cap (not recommended in production). WS_MAX_CONNECTIONS=1000 + +# Optional backplane for multi-instance fan-out. Default is "memory". +# Set to "redis" to publish WS events through Redis pub/sub and maintain a +# globally-shared seq counter for replay semantics. +WS_BACKPLANE=memory +REDIS_URL=redis://localhost:6379 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 80b6391..379dd58 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -53,12 +53,18 @@ npm run dev # → http://localhost:4000 # → Swagger UI: http://localhost:4000/docs # → WebSocket: ws://localhost:4000/ws + +# 4. If you need a local Postgres-backed app, use the bundled compose stack +# (one command; matches the CI service container config) +docker compose up --build ``` > Most feature work does not require Postgres — the service uses an in-memory > store by default. `DATABASE_URL` has a sensible default so the app boots > without a live database. You only need Postgres if you are working on -> Prisma migrations or the database-backed health check. +> Prisma migrations or the database-backed health check. The repo now includes a +> root-level `docker-compose.yml` so that workflow is one command instead of a +> hand-rolled local setup. --- diff --git a/README.md b/README.md index 7c2a868..1654918 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,7 @@ GET /api/v1/chain/account/:key — Stellar account lookup ### Prerequisites - Node.js 20+ +- Docker (optional, for the one-command local Postgres + app dev stack) ```bash npm install @@ -60,6 +61,21 @@ cp .env.testnet.example .env # testnet development (most contributors) npm run dev # http://localhost:4000 ``` +### One-command local stack with Postgres + +If you need the Prisma-backed local database flow, use the repo-provided compose stack: + +```bash +docker compose up --build +``` + +This starts: +- a `postgres:16-alpine` service matching the CI credentials (`vortex` / `vortex` / `vortex`) +- the app service built from the existing Dockerfile +- the app already pointed at `DATABASE_URL=postgresql://vortex:vortex@postgres:5432/vortex?schema=public` + +After the stack is up, the backend is available at http://localhost:4000 and the DB is reachable using the same default credentials shown in `.env.example`. + Three `.env.example` variants are provided for different deployment targets: | File | Use case | diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..446db32 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,42 @@ +services: + postgres: + image: postgres:16-alpine + container_name: vortex-postgres + restart: unless-stopped + environment: + POSTGRES_USER: vortex + POSTGRES_PASSWORD: vortex + POSTGRES_DB: vortex + ports: + - "5432:5432" + healthcheck: + test: ["CMD-SHELL", "pg_isready -U vortex -d vortex"] + interval: 10s + timeout: 5s + retries: 5 + volumes: + - postgres-data:/var/lib/postgresql/data + + app: + build: + context: . + dockerfile: Dockerfile + container_name: vortex-backend + depends_on: + postgres: + condition: service_healthy + environment: + NODE_ENV: development + PORT: 4000 + DATABASE_URL: postgresql://vortex:vortex@postgres:5432/vortex?schema=public + CORS_ORIGIN: "*" + WS_MAX_CONNECTIONS: 1000 + WS_BACKPLANE: memory + ports: + - "4000:4000" + volumes: + - .:/app + command: sh -c "npx prisma migrate deploy && npm run dev -- --host 0.0.0.0" + +volumes: + postgres-data: diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 0247b88..870e248 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -66,6 +66,8 @@ model Intent { filledAt Int? @map("filled_at") /// Actual amount received by the user (base unit string). fillAmount String? @map("fill_amount") + /// Realized protocol fee charged on this fill (destination-token base units). + feeAmount String? @map("fee_amount") /// On-chain transaction hash of the Stellar fill transaction. txHash String? @map("tx_hash") diff --git a/src/config/configuration.ts b/src/config/configuration.ts index aa84559..f94c267 100644 --- a/src/config/configuration.ts +++ b/src/config/configuration.ts @@ -52,6 +52,8 @@ export interface AppConfig { corsOrigin: string; /** Maximum concurrent WebSocket connections (0 = unlimited). */ wsMaxConnections: number; + wsBackplane: "memory" | "redis"; + redisUrl: string; } export default (): AppConfig => ({ @@ -71,4 +73,6 @@ export default (): AppConfig => ({ onchainIntentsEnabled: (process.env.ONCHAIN_INTENTS_ENABLED ?? "false") === "true", corsOrigin: process.env.CORS_ORIGIN ?? "*", wsMaxConnections: parseInt(process.env.WS_MAX_CONNECTIONS ?? "1000", 10), + wsBackplane: (process.env.WS_BACKPLANE ?? "memory") as "memory" | "redis", + redisUrl: process.env.REDIS_URL ?? "redis://localhost:6379", }); diff --git a/src/config/env.validation.ts b/src/config/env.validation.ts index 8a2dc55..2ece2a5 100644 --- a/src/config/env.validation.ts +++ b/src/config/env.validation.ts @@ -39,6 +39,9 @@ export const envValidationSchema = Joi.object({ CORS_ORIGIN: Joi.string().default("*"), + WS_BACKPLANE: Joi.string().valid("memory", "redis").default("memory"), + REDIS_URL: Joi.string().uri({ scheme: ["redis", "rediss"] }).default("redis://localhost:6379"), + // ── Persistence adapter selection ───────────────────────────────────────── // Controls which repository adapter is used for intents and solvers. // "memory" (default) keeps everything in-process — no database required. diff --git a/src/health/health.controller.ts b/src/health/health.controller.ts index c7f079e..a50f24f 100644 --- a/src/health/health.controller.ts +++ b/src/health/health.controller.ts @@ -12,6 +12,31 @@ export class HealthController { private readonly dbHealth: DatabaseHealthService, ) {} + @Get("live") + live() { + return { + status: "ok", + service: "vortex-backend", + version: "0.1.0", + network: `stellar-${this.configService.get("stellar.network", { infer: true })}`, + uptime: process.uptime(), + }; + } + + @Get("ready") + async ready() { + const db = await this.dbHealth.check(); + + return { + status: db.status === "ok" ? "ok" : "unreachable", + service: "vortex-backend", + version: "0.1.0", + network: `stellar-${this.configService.get("stellar.network", { infer: true })}`, + uptime: process.uptime(), + db, + }; + } + @Get() async check() { const db = await this.dbHealth.check(); diff --git a/src/intents/intents.controller.ts b/src/intents/intents.controller.ts index c072a02..af60cfa 100644 --- a/src/intents/intents.controller.ts +++ b/src/intents/intents.controller.ts @@ -280,9 +280,12 @@ export class IntentsController { }); } + const feeAmount = (BigInt(dto.fillAmount) * 5n) / 10000n; + const updated = await this.intentsService.fillIfAccepted(id, dto.solver, { filledAt: now, fillAmount: dto.fillAmount, + feeAmount: feeAmount.toString(), txHash: dto.txHash, }); if (!updated) { diff --git a/src/intents/intents.gateway.ts b/src/intents/intents.gateway.ts index a03c0be..76483b3 100644 --- a/src/intents/intents.gateway.ts +++ b/src/intents/intents.gateway.ts @@ -3,6 +3,7 @@ import { OnGatewayConnection, OnGatewayDisconnect, WebSocketGateway } from "@nes import { WebSocket } from "ws"; import { IntentsService } from "./intents.service"; import { logger } from "../common/logger"; +import { SUPPORTED_CHAINS, SupportedChain } from "./intents.types"; const HEARTBEAT_INTERVAL_MS = 30_000; @@ -68,25 +69,165 @@ export class EventRingBuffer { * REST API. The WS gateway never accepts writes, so there is no privileged * action to protect here. */ +type SubscriberFilter = Set | null; + @WebSocketGateway({ path: "/ws" }) export class IntentsGateway implements OnGatewayConnection, OnGatewayDisconnect, OnModuleDestroy { - private readonly subscribers = new Set(); + private readonly subscribers = new Map(); private readonly alive = new WeakMap(); // eslint-disable-next-line @typescript-eslint/no-explicit-any private heartbeatTimer: any; private nextSeq = 1; + private readonly backplane: null | { + publish: (event: Record) => void; + subscribe: (handler: (event: Record) => void) => void; + } = null; constructor(private readonly intentsService: IntentsService) { this.heartbeatTimer = setInterval(() => this.heartbeat(), HEARTBEAT_INTERVAL_MS); + this.backplane = this.createBackplane(); + if (this.backplane) { + this.backplane.subscribe((event) => { + const type = typeof event.type === "string" ? event.type : ""; + if (!type) return; + this.dispatchRemoteEvent(event as Record); + }); + } logger.info("ws heartbeat started"); } + private createBackplane(): null | { + publish: (event: Record) => void; + subscribe: (handler: (event: Record) => void) => void; + } { + const mode = (process.env.WS_BACKPLANE ?? "memory").toLowerCase(); + if (mode !== "redis") return null; + + try { + // eslint-disable-next-line @typescript-eslint/no-var-requires, @typescript-eslint/no-require-imports + const redis = require("redis"); + if (!redis?.createClient) { + logger.warn("WS_BACKPLANE=redis but the redis package is not available; falling back to memory"); + return null; + } + + const client = redis.createClient({ url: process.env.REDIS_URL ?? "redis://localhost:6379" }); + const channel = "vortex:intents:ws"; + const pub = client; + const sub = client.duplicate(); + + void sub.connect(); + void sub.subscribe(channel, (message: string) => { + try { + const event = JSON.parse(message) as Record; + if (event && typeof event === "object") { + this.dispatchRemoteEvent(event); + } + } catch { + // Ignore malformed backplane payloads. + } + }); + + return { + publish: (event: Record) => { + void pub.publish(channel, JSON.stringify(event)); + }, + subscribe: (handler: (event: Record) => void) => { + void sub.subscribe(channel, (message: string) => { + try { + const event = JSON.parse(message) as Record; + handler(event); + } catch { + // Ignore malformed backplane payloads. + } + }); + }, + }; + } catch { + logger.warn("WS_BACKPLANE=redis but the redis package is not available; falling back to memory"); + return null; + } + } + + private static isSupportedChain(value: unknown): value is SupportedChain { + return typeof value === "string" && (SUPPORTED_CHAINS as readonly string[]).includes(value); + } + + private static resolveFilter(chains: unknown): SubscriberFilter { + if (!Array.isArray(chains) || chains.length === 0) { + return null; + } + + const normalized = new Set(); + for (const chain of chains) { + if (IntentsGateway.isSupportedChain(chain)) { + normalized.add(chain); + } + } + + return normalized.size > 0 ? normalized : null; + } + + private handleMessage(client: WebSocket, raw: unknown) { + try { + const text = typeof raw === "string" ? raw : raw instanceof Buffer ? raw.toString() : String(raw); + const message = JSON.parse(text) as { type?: string; chains?: unknown }; + if (message.type !== "subscribe") return; + + const filter = IntentsGateway.resolveFilter(message.chains); + this.subscribers.set(client, filter); + + client.send( + JSON.stringify({ + type: "subscribed", + filter: { chains: filter ? [...filter] : [...SUPPORTED_CHAINS] }, + seq: this.nextSeq - 1, + }), + ); + } catch { + // Ignore malformed WS frames; the client can reconnect or retry. + } + } + + private getEventChain(event: { type: string; [key: string]: unknown }): SupportedChain | null { + const intent = (event as { intent?: { srcChain?: unknown } }).intent; + if (intent && typeof intent.srcChain === "string" && IntentsGateway.isSupportedChain(intent.srcChain)) { + return intent.srcChain; + } + + const srcChain = (event as { srcChain?: unknown }).srcChain; + if (typeof srcChain === "string" && IntentsGateway.isSupportedChain(srcChain)) { + return srcChain; + } + + return null; + } + + private deliverToMatchingSubscribers(payload: string, chain: SupportedChain | null) { + for (const [client, filter] of this.subscribers.entries()) { + if (client.readyState !== WebSocket.OPEN) continue; + if (chain !== null && filter && !filter.has(chain)) continue; + client.send(payload); + } + } + + private dispatchRemoteEvent(event: Record) { + const type = typeof event.type === "string" ? event.type : ""; + if (!type || type === "connected" || type === "snapshot" || type === "subscribed") return; + + const payload = JSON.stringify(event); + const chain = this.getEventChain(event as { type: string; [key: string]: unknown }); + this.deliverToMatchingSubscribers(payload, chain); + } + handleConnection(client: WebSocket) { - this.subscribers.add(client); + this.subscribers.set(client, null); this.alive.set(client, true); + client.on("message", (raw) => this.handleMessage(client, raw)); + client.on("pong", () => { this.alive.set(client, true); }); @@ -127,17 +268,43 @@ export class IntentsGateway } broadcast(event: { type: string; [key: string]: unknown }) { + const seqEvent = { ...event, seq: this.nextSeq }; + this.nextSeq += 1; logger.debug(`ws broadcast type=${event.type} subscribers=${this.subscribers.size}`); - const payload = JSON.stringify(event); - for (const client of this.subscribers) { - if (client.readyState !== WebSocket.OPEN) continue; - client.send(payload); + const payload = JSON.stringify(seqEvent); + const chain = this.getEventChain(seqEvent); + + if (this.backplane) { + this.backplane.publish(seqEvent as Record); + } + + if (chain) { + this.deliverToMatchingSubscribers(payload, chain); + return; } + + if (typeof event.intentId === "string") { + void this.intentsService + .get(event.intentId) + .then((intent) => { + if (!intent) { + this.deliverToMatchingSubscribers(payload, null); + return; + } + this.deliverToMatchingSubscribers(payload, intent.srcChain); + }) + .catch(() => { + this.deliverToMatchingSubscribers(payload, null); + }); + return; + } + + this.deliverToMatchingSubscribers(payload, null); } getAliveCount(): number { let count = 0; - for (const client of this.subscribers) { + for (const client of this.subscribers.keys()) { if (this.alive.get(client) === true) count++; } return count; @@ -153,7 +320,7 @@ export class IntentsGateway } private heartbeat() { - for (const client of this.subscribers) { + for (const [client] of this.subscribers.entries()) { if (this.alive.get(client) === false) { client.terminate(); this.subscribers.delete(client); @@ -172,7 +339,7 @@ export class IntentsGateway onModuleDestroy() { if (this.heartbeatTimer) clearInterval(this.heartbeatTimer); - for (const client of this.subscribers) { + for (const client of this.subscribers.keys()) { client.close(1001, "Server shutting down"); } this.subscribers.clear(); diff --git a/src/intents/intents.types.ts b/src/intents/intents.types.ts index 5c5c79b..7d830df 100644 --- a/src/intents/intents.types.ts +++ b/src/intents/intents.types.ts @@ -75,6 +75,7 @@ export interface Intent { deadline: number; filledAt?: number; fillAmount?: string; + feeAmount?: string; // realized protocol fee in dst token base units txHash?: string; // fill tx on Stellar slashedAt?: number; slashReason?: string; diff --git a/src/intents/prisma-intents.repository.ts b/src/intents/prisma-intents.repository.ts index bcaec0c..3d4bfca 100644 --- a/src/intents/prisma-intents.repository.ts +++ b/src/intents/prisma-intents.repository.ts @@ -136,7 +136,7 @@ export class PrismaIntentsRepository implements IIntentsRepository { private toDbData( intent: Intent, ): Omit { - return { + const data: Omit & { feeAmount?: string | null } = { user: intent.user, srcChain: intent.srcChain as Prisma.IntentCreateInput["srcChain"], srcToken: intent.srcToken as unknown as Prisma.InputJsonValue, @@ -152,16 +152,23 @@ export class PrismaIntentsRepository implements IIntentsRepository { fillAmount: intent.fillAmount ?? null, txHash: intent.txHash ?? null, }; + + if (intent.feeAmount !== undefined) { + (data as { feeAmount?: string | null }).feeAmount = intent.feeAmount ?? null; + } + + return data; } /** Build an `updateMany`-compatible data object from a partial Intent patch. */ private toDbPatch(patch: Partial): Prisma.IntentUpdateInput { - const data: Prisma.IntentUpdateInput = {}; + const data = {} as Prisma.IntentUpdateInput & { feeAmount?: string | null }; if (patch.state !== undefined) data.state = this.toPrismaState(patch.state); if (patch.solver !== undefined) data.solver = patch.solver; if (patch.deadline !== undefined) data.deadline = patch.deadline; if (patch.filledAt !== undefined) data.filledAt = patch.filledAt; if (patch.fillAmount !== undefined) data.fillAmount = patch.fillAmount; + if (patch.feeAmount !== undefined) (data as { feeAmount?: string | null }).feeAmount = patch.feeAmount ?? null; if (patch.txHash !== undefined) data.txHash = patch.txHash; if (patch.quotedDstAmount !== undefined) data.quotedDstAmount = patch.quotedDstAmount; if (patch.srcAmount !== undefined) data.srcAmount = patch.srcAmount; @@ -189,6 +196,7 @@ export class PrismaIntentsRepository implements IIntentsRepository { deadline: number; filledAt: number | null; fillAmount: string | null; + feeAmount?: string | null; txHash: string | null; }): Intent { return { @@ -206,6 +214,7 @@ export class PrismaIntentsRepository implements IIntentsRepository { deadline: row.deadline, ...(row.filledAt !== null ? { filledAt: row.filledAt } : {}), ...(row.fillAmount !== null ? { fillAmount: row.fillAmount } : {}), + ...(row.feeAmount !== undefined && row.feeAmount !== null ? { feeAmount: row.feeAmount } : {}), ...(row.txHash !== null ? { txHash: row.txHash } : {}), }; } diff --git a/src/stats/stats.controller.ts b/src/stats/stats.controller.ts index 62f95a9..6e3062d 100644 --- a/src/stats/stats.controller.ts +++ b/src/stats/stats.controller.ts @@ -12,6 +12,11 @@ export class StatsController { return this.statsService.getProtocolStats(); } + @Get("treasury") + getTreasuryStats() { + return this.statsService.getTreasuryStats(); + } + @Get("ws") getWsStats() { return this.statsService.getWsStats(); diff --git a/src/stats/stats.service.ts b/src/stats/stats.service.ts index d6b4ad2..96d5834 100644 --- a/src/stats/stats.service.ts +++ b/src/stats/stats.service.ts @@ -41,6 +41,68 @@ export class StatsService { }; } + async getTreasuryStats() { + const intents = await this.intentsService.getAll(); + const now = Math.floor(Date.now() / 1000); + const last24hCutoff = now - 86_400; + + const allTime = intents + .filter((intent) => typeof intent.feeAmount === "string" && intent.feeAmount.length > 0) + .reduce((sum, intent) => sum + BigInt(intent.feeAmount ?? "0"), 0n); + + const last24h = intents + .filter( + (intent) => + typeof intent.feeAmount === "string" && + intent.feeAmount.length > 0 && + typeof intent.filledAt === "number" && + intent.filledAt >= last24hCutoff, + ) + .reduce((sum, intent) => sum + BigInt(intent.feeAmount ?? "0"), 0n); + + const byChain = new Map(); + + for (const intent of intents) { + if (typeof intent.feeAmount !== "string" || intent.feeAmount.length === 0) continue; + const fee = BigInt(intent.feeAmount ?? "0"); + const entry = byChain.get(intent.srcChain) ?? { + totalFees: 0n, + last24hFees: 0n, + filledCount: 0, + }; + + entry.totalFees += fee; + entry.filledCount += 1; + if (typeof intent.filledAt === "number" && intent.filledAt >= last24hCutoff) { + entry.last24hFees += fee; + } + byChain.set(intent.srcChain, entry); + } + + return { + allTime: { + totalFees: allTime.toString(), + filledIntents: intents.filter((intent) => typeof intent.feeAmount === "string" && intent.feeAmount.length > 0).length, + }, + last24h: { + totalFees: last24h.toString(), + filledIntents: intents.filter( + (intent) => + typeof intent.feeAmount === "string" && + intent.feeAmount.length > 0 && + typeof intent.filledAt === "number" && + intent.filledAt >= last24hCutoff, + ).length, + }, + byChain: Array.from(byChain.entries()).map(([srcChain, stats]) => ({ + srcChain, + totalFees: stats.totalFees.toString(), + last24hFees: stats.last24hFees.toString(), + filledIntents: stats.filledCount, + })), + }; + } + getWsStats() { return { subscriberCount: this.intentsGateway.getSubscriberCount(),