diff --git a/.env.example b/.env.example index 1d8ed5ef1..076234c4f 100644 --- a/.env.example +++ b/.env.example @@ -42,6 +42,13 @@ START_LEDGER= # Polling interval in ms (default: 6000 = ~1 ledger) POLL_INTERVAL_MS=6000 +# Optional candles read API and OHLC refresh worker. Disabled when unset or 0. +# Enable only after usable single-network OHLC tables/functions are installed. +# Use an integer interval in ms (1..2147483647), for example 60000. +# Invalid values or SKIP_INDEXER=true disable both; NETWORKS must contain only +# STELLAR_NETWORK. Generate OpenAPI with these same deployment settings. +OHLC_REFRESH_INTERVAL_MS=0 + # ─── SAC Contract IDs ──────────────────────────────────────── # Comma-separated list of Soroban SAC (Stellar Asset Contract) IDs to watch. # When unset, defaults to the native XLM SAC for the configured STELLAR_NETWORK: diff --git a/README.md b/README.md index d6028b71a..76f50c6cb 100644 --- a/README.md +++ b/README.md @@ -684,6 +684,34 @@ curl -OJ "http://localhost:3000/transfers.parquet?fromLedger=5840000&toLedger=58 *** +### `GET /candles/:bucket/:contractId` + +Returns OHLC candles from the existing `ohlc` aggregate tables. `bucket` is +`1m`, `1h`, or `1d`; `contractId` is a 56-character Stellar contract address. +Pagination uses `limit` (1–1000, default 100) and `offset` (non-negative, default 0). +Refreshes are owned by the worker; there is no public refresh endpoint. + +The database must already contain working `ohlc.candles_*` tables and +`ohlc.refresh_candles_*()` functions. They are not installed by Prisma's current +startup path; the legacy `sql/001_ohlc_aggregates.sql` requires separate repair +and deployment. These legacy aggregates do not separate networks, so they +must not be used with a database containing transfers from multiple networks. + +The read route and scheduled refresh share one opt-in: set +`OHLC_REFRESH_INTERVAL_MS=60000` to enable both after those database prerequisites +are met. Unset, invalid, or `0` disables both; `SKIP_INDEXER=true` also disables +both. They require exactly one enabled network matching `STELLAR_NETWORK` +(default `testnet`); multi-network configurations leave the route unmounted. +The worker stops scheduling on shutdown and skips ticks while an earlier +refresh is still running. + +Run `npm run docs:openapi` with the same environment as the deployment. Without +the opt-in, the generated specification omits the candles route. For an enabled +single-network deployment, generate it with `OHLC_REFRESH_INTERVAL_MS=60000 npm +run docs:openapi` (and the deployment's network settings). + +*** + ## Environment Variables | Variable | Default | Description | @@ -697,6 +725,7 @@ curl -OJ "http://localhost:3000/transfers.parquet?fromLedger=5840000&toLedger=58 | `HORIZON_EVENTS_PATH` | `/events` | Horizon contract-events path used by the fallback source. | | `START_LEDGER` | *(tip)* | Ledger to start indexing from. Leave blank to resume from DB state or start near the tip. | | `POLL_INTERVAL_MS` | `6000` | Polling interval in ms (\~1 ledger ≈ 6 s) | +| `OHLC_REFRESH_INTERVAL_MS` | disabled | Enables the candles read route and worker with a positive integer interval in ms (max 2147483647). Requires working single-network OHLC aggregates. Invalid, unset, `0`, or `SKIP_INDEXER=true` disables both. | | `CONTRACT_IDS` | *(all)* | Comma-separated token contract IDs to watch. Empty = watch all (very heavy on mainnet) | | `EVENTS_BATCH_SIZE` | `10000` | Max events per RPC call (Stellar RPC hard-cap is 10 000) | | `RETENTION_DAYS` | `30` | Delete transfers older than N days (keeps DB within free-tier limits) | diff --git a/docker-compose.yml b/docker-compose.yml index ffdf83ad6..03cd53e5b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -53,6 +53,7 @@ services: NETWORKS: ${NETWORKS:-} START_LEDGER: ${START_LEDGER:-} POLL_INTERVAL_MS: ${POLL_INTERVAL_MS:-6000} + OHLC_REFRESH_INTERVAL_MS: ${OHLC_REFRESH_INTERVAL_MS:-0} SAC_CONTRACT_IDS: ${SAC_CONTRACT_IDS:-} SAC_CONTRACT_IDS_TESTNET: ${SAC_CONTRACT_IDS_TESTNET:-} SAC_CONTRACT_IDS_MAINNET: ${SAC_CONTRACT_IDS_MAINNET:-} diff --git a/src/__tests__/ohlcOpenapi.test.ts b/src/__tests__/ohlcOpenapi.test.ts new file mode 100644 index 000000000..66a1678f4 --- /dev/null +++ b/src/__tests__/ohlcOpenapi.test.ts @@ -0,0 +1,63 @@ +describe("OHLC OpenAPI opt-in", () => { + const originalEnv = { ...process.env }; + + beforeEach(() => { + jest.clearAllMocks(); + delete process.env.OHLC_REFRESH_INTERVAL_MS; + delete process.env.SKIP_INDEXER; + process.env.NETWORKS = "testnet"; + process.env.STELLAR_NETWORK = "testnet"; + jest.spyOn(console, "log").mockImplementation(() => {}); + }); + + afterEach(() => { + process.env = { ...originalEnv }; + jest.restoreAllMocks(); + }); + + function generatePaths(): Record { + let paths: Record = {}; + jest.isolateModules(() => { + paths = require("../openapi/build").buildOpenApiDocument().paths; + }); + return paths; + } + + it.each([undefined, "", "0", "-1", "1.5", "60000ms", "2147483648"])( + "omits candles without a valid opt-in (%s)", + (interval) => { + if (interval !== undefined) process.env.OHLC_REFRESH_INTERVAL_MS = interval; + + const paths = generatePaths(); + + expect(paths).not.toHaveProperty("/candles/{bucket}/{contractId}"); + expect(paths).not.toHaveProperty("/candles/refresh"); + }, + ); + + it("documents only the read route when enabled", () => { + process.env.OHLC_REFRESH_INTERVAL_MS = "60000"; + + const paths = generatePaths(); + + expect(paths["/candles/{bucket}/{contractId}"]).toHaveProperty("get"); + expect(paths).not.toHaveProperty("/candles/refresh"); + }); + + it("omits candles when API-only mode disables the worker", () => { + process.env.OHLC_REFRESH_INTERVAL_MS = "60000"; + process.env.SKIP_INDEXER = "true"; + + expect(generatePaths()).not.toHaveProperty("/candles/{bucket}/{contractId}"); + }); + + it.each(["testnet,mainnet", "mainnet"])( + "omits candles for an incompatible network configuration (%s)", + (networks) => { + process.env.OHLC_REFRESH_INTERVAL_MS = "60000"; + process.env.NETWORKS = networks; + + expect(generatePaths()).not.toHaveProperty("/candles/{bucket}/{contractId}"); + }, + ); +}); diff --git a/src/__tests__/ohlcRefreshWorker.test.ts b/src/__tests__/ohlcRefreshWorker.test.ts new file mode 100644 index 000000000..1d0154d84 --- /dev/null +++ b/src/__tests__/ohlcRefreshWorker.test.ts @@ -0,0 +1,123 @@ +jest.mock("../db", () => ({ prisma: { $queryRaw: jest.fn() } })); + +import { prisma } from "../db"; +import { refreshOhlcAggregates, startOhlcRefreshWorker } from "../workers/ohlc-refresh"; + +const mockQueryRaw = prisma.$queryRaw as jest.Mock; + +function pendingQuery() { + let resolve!: (rows: Array<{ rows_inserted: number; rows_updated: number }>) => void; + const promise = new Promise>((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +describe("OHLC refresh worker", () => { + beforeEach(() => { + jest.useFakeTimers(); + mockQueryRaw.mockReset().mockResolvedValue([]); + jest.spyOn(console, "log").mockImplementation(() => {}); + jest.spyOn(console, "error").mockImplementation(() => {}); + }); + + afterEach(() => { + jest.clearAllTimers(); + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + it("refreshes all three buckets and preserves inserted and updated counts", async () => { + mockQueryRaw + .mockResolvedValueOnce([{ rows_inserted: 1, rows_updated: 2 }]) + .mockResolvedValueOnce([{ rows_inserted: 3, rows_updated: 4 }]) + .mockResolvedValueOnce([{ rows_inserted: 5, rows_updated: 6 }]); + + await expect(refreshOhlcAggregates()).resolves.toEqual({ + oneMinute: { inserted: 1, updated: 2 }, + oneHour: { inserted: 3, updated: 4 }, + oneDay: { inserted: 5, updated: 6 }, + duration_ms: 0, + }); + expect(mockQueryRaw).toHaveBeenCalledTimes(3); + const queries = mockQueryRaw.mock.calls.map(([parts]) => parts.join("")); + expect(queries[0]).toContain("ohlc.refresh_candles_1m()"); + expect(queries[1]).toContain("ohlc.refresh_candles_1h()"); + expect(queries[2]).toContain("ohlc.refresh_candles_1d()"); + }); + + it("reports zero counts when the refresh functions return no rows", async () => { + await expect(refreshOhlcAggregates()).resolves.toEqual({ + oneMinute: { inserted: 0, updated: 0 }, + oneHour: { inserted: 0, updated: 0 }, + oneDay: { inserted: 0, updated: 0 }, + duration_ms: 0, + }); + }); + + it("does not expose a database error in a thrown error or log", async () => { + mockQueryRaw.mockRejectedValueOnce( + new Error("connection failed: postgresql://admin:secret@private-db/wraith"), + ); + + await expect(refreshOhlcAggregates()).rejects.toThrow(/^OHLC refresh failed$/); + expect(console.error).not.toHaveBeenCalled(); + expect(console.log).not.toHaveBeenCalled(); + }); + + it("waits for the configured interval and stops scheduling when disposed", async () => { + const stop = startOhlcRefreshWorker(5_000); + + expect(mockQueryRaw).not.toHaveBeenCalled(); + await jest.advanceTimersByTimeAsync(4_999); + expect(mockQueryRaw).not.toHaveBeenCalled(); + await jest.advanceTimersByTimeAsync(1); + expect(mockQueryRaw).toHaveBeenCalledTimes(3); + + stop(); + stop(); + expect(jest.getTimerCount()).toBe(0); + await jest.advanceTimersByTimeAsync(10_000); + expect(mockQueryRaw).toHaveBeenCalledTimes(3); + }); + + it("skips overlapping ticks while a refresh is still running", async () => { + const pending = pendingQuery(); + mockQueryRaw.mockReturnValueOnce(pending.promise); + const stop = startOhlcRefreshWorker(1_000); + + await jest.advanceTimersByTimeAsync(1_000); + expect(mockQueryRaw).toHaveBeenCalledTimes(3); + await jest.advanceTimersByTimeAsync(3_000); + expect(mockQueryRaw).toHaveBeenCalledTimes(3); + + pending.resolve([]); + await jest.advanceTimersByTimeAsync(1_000); + expect(mockQueryRaw).toHaveBeenCalledTimes(6); + stop(); + }); + + it("waits for remaining bucket queries after a failure and then retries safely", async () => { + const pending = pendingQuery(); + mockQueryRaw + .mockRejectedValueOnce(new Error("postgresql://admin:secret@private-db/wraith")) + .mockReturnValueOnce(pending.promise); + const stop = startOhlcRefreshWorker(1_000); + + await jest.advanceTimersByTimeAsync(1_000); + await jest.advanceTimersByTimeAsync(3_000); + expect(mockQueryRaw).toHaveBeenCalledTimes(3); + expect(console.error).not.toHaveBeenCalled(); + + pending.resolve([]); + await jest.advanceTimersByTimeAsync(0); + expect(console.error).toHaveBeenCalledTimes(1); + expect(console.error).toHaveBeenCalledWith( + "[ohlc] Refresh failed; retrying on the next interval.", + ); + await jest.advanceTimersByTimeAsync(1_000); + expect(mockQueryRaw).toHaveBeenCalledTimes(6); + expect(console.log).toHaveBeenCalledTimes(1); + stop(); + }); +}); diff --git a/src/__tests__/ohlcStartup.test.ts b/src/__tests__/ohlcStartup.test.ts new file mode 100644 index 000000000..44fa8ffb4 --- /dev/null +++ b/src/__tests__/ohlcStartup.test.ts @@ -0,0 +1,119 @@ +const mockStopOhlcRefreshWorker = jest.fn(); +const mockStartOhlcRefreshWorker = jest.fn(() => mockStopOhlcRefreshWorker); +const mockStartAllIndexers = jest.fn(); +const mockDisconnect = jest.fn().mockResolvedValue(undefined); +const mockListen = jest.fn(); + +jest.mock("dotenv/config", () => ({})); +jest.mock("http", () => ({ createServer: () => ({ listen: mockListen }) })); +jest.mock("child_process", () => ({ execSync: jest.fn() })); +jest.mock("../api", () => ({ createApp: jest.fn() })); +jest.mock("../indexer", () => ({ startAllIndexers: mockStartAllIndexers })); +jest.mock("../db", () => ({ prisma: { $disconnect: mockDisconnect } })); +jest.mock("../ws", () => ({ attachWebSocketServer: jest.fn() })); +jest.mock("../graphql/subscriptions", () => ({ + attachGraphQLSubscriptions: jest.fn(), + SUBSCRIPTIONS_PATH: "/graphql/subscriptions", +})); +jest.mock("../workers/webhooks", () => ({ startWebhookWorker: jest.fn() })); +jest.mock("../jobs/retention", () => ({ startPartitionRetentionJob: jest.fn() })); +jest.mock("../tokenCache", () => ({ initTokenCache: jest.fn().mockResolvedValue(undefined) })); +jest.mock("../schemaGuard", () => ({ pushSchema: jest.fn(() => ({ ok: true })) })); +jest.mock("../workers/ohlc-refresh", () => ({ + startOhlcRefreshWorker: mockStartOhlcRefreshWorker, +})); + +describe("OHLC refresh startup wiring", () => { + const originalEnv = { ...process.env }; + const signalHandlers = new Map Promise>(); + + beforeEach(() => { + jest.clearAllMocks(); + delete process.env.OHLC_REFRESH_INTERVAL_MS; + delete process.env.SKIP_INDEXER; + process.env.NETWORKS = "testnet"; + process.env.STELLAR_NETWORK = "testnet"; + signalHandlers.clear(); + jest.spyOn(console, "log").mockImplementation(() => {}); + jest.spyOn(process, "exit").mockImplementation(() => undefined as never); + jest.spyOn(process, "on").mockImplementation((event, listener) => { + signalHandlers.set(event, listener as () => Promise); + return process; + }); + }); + + afterEach(() => { + process.env = { ...originalEnv }; + jest.restoreAllMocks(); + }); + + async function startService(): Promise { + jest.isolateModules(() => { + require("../index"); + }); + await new Promise((resolve) => setImmediate(resolve)); + } + + it.each([undefined, "", "0", "-1", "1.5", "60000ms", "1e3", "0x10", "Infinity", "2147483648"])( + "does not start the worker for a missing, disabled or invalid interval (%s)", + async (interval) => { + if (interval !== undefined) process.env.OHLC_REFRESH_INTERVAL_MS = interval; + + await startService(); + + expect(mockStartOhlcRefreshWorker).not.toHaveBeenCalled(); + expect(mockStartAllIndexers).toHaveBeenCalledTimes(1); + }, + ); + + it.each(["1", "60000", "2147483647", " 60000 "])( + "starts one worker with a valid interval (%s)", + async (interval) => { + process.env.OHLC_REFRESH_INTERVAL_MS = interval; + + await startService(); + + expect(mockStartOhlcRefreshWorker).toHaveBeenCalledTimes(1); + expect(mockStartOhlcRefreshWorker).toHaveBeenCalledWith(Number(interval)); + expect(mockStartAllIndexers).toHaveBeenCalledTimes(1); + }, + ); + + it("skips OHLC refresh in API-only mode even with a valid interval", async () => { + process.env.OHLC_REFRESH_INTERVAL_MS = "60000"; + process.env.SKIP_INDEXER = "true"; + + await startService(); + + expect(mockStartOhlcRefreshWorker).not.toHaveBeenCalled(); + expect(mockStartAllIndexers).not.toHaveBeenCalled(); + expect(mockListen).toHaveBeenCalledTimes(1); + }); + + it.each(["testnet,mainnet", "mainnet"])( + "skips legacy OHLC refresh when enabled networks disagree with the single default (%s)", + async (networks) => { + process.env.OHLC_REFRESH_INTERVAL_MS = "60000"; + process.env.NETWORKS = networks; + + await startService(); + + expect(mockStartOhlcRefreshWorker).not.toHaveBeenCalled(); + expect(mockStartAllIndexers).toHaveBeenCalledTimes(1); + }, + ); + + it.each(["SIGINT", "SIGTERM"])("stops OHLC refresh before disconnecting on %s", async (signal) => { + process.env.OHLC_REFRESH_INTERVAL_MS = "60000"; + await startService(); + + await signalHandlers.get(signal)!(); + + expect(mockStopOhlcRefreshWorker).toHaveBeenCalledTimes(1); + expect(mockDisconnect).toHaveBeenCalledTimes(1); + expect(mockStopOhlcRefreshWorker.mock.invocationCallOrder[0]).toBeLessThan( + mockDisconnect.mock.invocationCallOrder[0], + ); + expect(process.exit).toHaveBeenCalledWith(0); + }); +}); diff --git a/src/api.ts b/src/api.ts index cbd35182b..d0dd4d29d 100644 --- a/src/api.ts +++ b/src/api.ts @@ -8,6 +8,8 @@ import { getLatestLedger } from "./rpc"; import { getIndexerStats, getAllIndexerStats, runningNetworks } from "./indexer"; import { currentNetwork, enabledNetworks, type Network } from "./network"; import { createAccountsRouter } from "./api/accounts"; +import { createCandlesRouter } from "./api/candles"; +import { getOhlcRefreshIntervalMs } from "./ohlcConfig"; import { createWebhooksRouter } from "./api/webhooks"; import { createLinqWebhookRouter } from "./api/linqWebhook"; import { createOfframpRouter } from "./api/offramp"; @@ -247,6 +249,9 @@ export function createApp(): express.Application { // ─── Accounts routes ───────────────────────────────────────────────────── app.use("/accounts", createAccountsRouter()); + if (getOhlcRefreshIntervalMs() !== undefined) { + app.use("/candles", createCandlesRouter()); + } // ─── Webhook subscription management ───────────────────────────────────── app.use("/webhooks", createWebhooksRouter()); diff --git a/src/api/candles.ts b/src/api/candles.ts index fc7e6be71..c4248f539 100644 --- a/src/api/candles.ts +++ b/src/api/candles.ts @@ -1,5 +1,7 @@ import { Router } from "express"; import { prisma } from "../db"; +import { candlesParamsSchema, candlesQuerySchema } from "../openapi/schemas"; +import { parseOr400 } from "../openapi/validation"; export interface Candle { timeBucket: string; @@ -47,7 +49,9 @@ async function queryCandlesFromAggregate( ORDER BY time_bucket DESC LIMIT $2 OFFSET $3 `, - [contractId, limit, offset], + contractId, + limit, + offset, ); return rows.map((row) => ({ @@ -70,79 +74,26 @@ export function createCandlesRouter(): Router { router.get("/:bucket/:contractId", async (req, res) => { try { - const { bucket, contractId } = req.params; - const limit = Math.min(parseInt(req.query.limit as string) || 100, 1000); - const offset = parseInt(req.query.offset as string) || 0; - - if (!["1m", "1h", "1d"].includes(bucket)) { - return res - .status(400) - .json({ error: "Invalid bucket: must be 1m, 1h, or 1d" }); - } - - if (!contractId.match(/^C[A-Z2-7]{55}$/)) { - return res.status(400).json({ error: "Invalid contract ID format" }); - } + const params = parseOr400(candlesParamsSchema, req.params, res); + if (!params) return; + const query = parseOr400(candlesQuerySchema, req.query, res); + if (!query) return; + const { bucket, contractId } = params; + const { limit, offset } = query; const candles = await queryCandlesFromAggregate( - bucket as "1m" | "1h" | "1d", + bucket, contractId, limit, offset, ); res.json({ bucket, contractId, candles }); - } catch (err) { - console.error("[candles] Query failed:", err); + } catch { + console.error("[candles] Query failed"); res.status(500).json({ error: "Failed to query candles" }); } }); - router.post("/refresh", async (req, res) => { - try { - const [result1m, result1h, result1d] = await Promise.all([ - prisma.$queryRaw< - Array<{ rows_inserted: number; rows_updated: number }> - >` - SELECT rows_inserted, rows_updated FROM ohlc.refresh_candles_1m() - `, - prisma.$queryRaw< - Array<{ rows_inserted: number; rows_updated: number }> - >` - SELECT rows_inserted, rows_updated FROM ohlc.refresh_candles_1h() - `, - prisma.$queryRaw< - Array<{ rows_inserted: number; rows_updated: number }> - >` - SELECT rows_inserted, rows_updated FROM ohlc.refresh_candles_1d() - `, - ]); - - res.json({ - oneMinute: result1m[0] - ? { - inserted: result1m[0].rows_inserted, - updated: result1m[0].rows_updated, - } - : { inserted: 0, updated: 0 }, - oneHour: result1h[0] - ? { - inserted: result1h[0].rows_inserted, - updated: result1h[0].rows_updated, - } - : { inserted: 0, updated: 0 }, - oneDay: result1d[0] - ? { - inserted: result1d[0].rows_inserted, - updated: result1d[0].rows_updated, - } - : { inserted: 0, updated: 0 }, - }); - } catch (err) { - console.error("[candles] Refresh failed:", err); - res.status(500).json({ error: "Failed to refresh candles" }); - } - }); - return router; } diff --git a/src/db.ts b/src/db.ts index ea9664919..56648333a 100644 --- a/src/db.ts +++ b/src/db.ts @@ -26,10 +26,9 @@ export const prisma = globalForPrisma.prisma ?? withReadReplicas( new PrismaClient({ - log: - process.env.NODE_ENV === "development" - ? ["query", "warn", "error"] - : ["warn", "error"], + // Preserve operational warnings; handle failures at the application + // boundary without logging raw database errors or query details. + log: ["warn"], }), { replicaUrls } ); diff --git a/src/index.ts b/src/index.ts index 5a73c0df2..8eebc89e9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,6 +11,8 @@ import { startPartitionRetentionJob } from "./jobs/retention"; import { initTokenCache } from "./tokenCache"; import { enabledNetworks } from "./network"; import { onlyAddsUniqueConstraints, pushSchema } from "./schemaGuard"; +import { startOhlcRefreshWorker } from "./workers/ohlc-refresh"; +import { getOhlcRefreshIntervalMs } from "./ohlcConfig"; const PORT = parseInt(process.env.PORT ?? "3000", 10); @@ -78,8 +80,10 @@ async function main() { console.log("[wraith] Database ready."); // ── Graceful shutdown ────────────────────────────────────────────────────── + let stopOhlcRefreshWorker: (() => void) | undefined; const shutdown = async (signal: string) => { console.log(`\n[wraith] Received ${signal} — shutting down gracefully…`); + stopOhlcRefreshWorker?.(); await prisma.$disconnect(); process.exit(0); }; @@ -113,6 +117,12 @@ async function main() { // displayAmount values would silently fall back to 7 decimals for every token. await Promise.all(enabledNetworks().map((network) => initTokenCache(network))); + // Opt in only after the OHLC tables and refresh functions are available. + const ohlcIntervalMs = getOhlcRefreshIntervalMs(); + if (ohlcIntervalMs !== undefined) { + stopOhlcRefreshWorker = startOhlcRefreshWorker(ohlcIntervalMs); + } + // ── Start indexer in the background ─────────────────────────────────────── // startIndexer() runs an infinite loop; we intentionally don't await it // so the API stays responsive while indexing happens concurrently. diff --git a/src/ohlcConfig.ts b/src/ohlcConfig.ts new file mode 100644 index 000000000..0a400a46a --- /dev/null +++ b/src/ohlcConfig.ts @@ -0,0 +1,15 @@ +import { currentNetwork, enabledNetworks } from "./network"; + +/** Shared opt-in for the legacy single-network candles API and refresh worker. */ +export function getOhlcRefreshIntervalMs(): number | undefined { + if (process.env.SKIP_INDEXER === "true") return undefined; + const networks = enabledNetworks(); + if (networks.length !== 1 || networks[0] !== currentNetwork()) return undefined; + + const raw = process.env.OHLC_REFRESH_INTERVAL_MS?.trim() ?? ""; + const interval = Number(raw); + // Node clamps delays above 2^31-1 to 1ms, which would overload the database. + if (!/^\d+$/.test(raw) || !Number.isSafeInteger(interval) + || interval <= 0 || interval > 2_147_483_647) return undefined; + return interval; +} diff --git a/src/openapi/build.ts b/src/openapi/build.ts index 45a445653..61fbd6f5c 100644 --- a/src/openapi/build.ts +++ b/src/openapi/build.ts @@ -1,9 +1,13 @@ import { mkdirSync, writeFileSync } from "fs"; import path from "path"; import { OpenAPIRegistry, OpenApiGeneratorV3 } from "@asteasolutions/zod-to-openapi"; +import { getOhlcRefreshIntervalMs } from "../ohlcConfig"; import { addressPathSchema, booleanOkResponseSchema, + candlesParamsSchema, + candlesQuerySchema, + candlesResponseSchema, errorResponseSchema, healthzResponseSchema, hostFnQuerySchema, @@ -515,6 +519,22 @@ registry.registerPath({ }, }); +if (getOhlcRefreshIntervalMs() !== undefined) { + registry.registerPath({ + method: "get", + path: "/candles/{bucket}/{contractId}", + summary: "OHLC candles for a token contract (single-network deployments only)", + request: { + params: candlesParamsSchema, + query: candlesQuerySchema, + }, + responses: { + 200: { description: "OK", content: { "application/json": { schema: candlesResponseSchema } } }, + ...commonErrorResponses, + }, + }); +} + const generator = new OpenApiGeneratorV3(registry.definitions); const document = generator.generateDocument({ openapi: "3.0.3", diff --git a/src/workers/ohlc-refresh.ts b/src/workers/ohlc-refresh.ts index a23b7d261..e387271d2 100644 --- a/src/workers/ohlc-refresh.ts +++ b/src/workers/ohlc-refresh.ts @@ -11,7 +11,9 @@ export async function refreshOhlcAggregates(): Promise { const start = Date.now(); try { - const [result1m, result1h, result1d] = await Promise.all([ + // Wait for every bucket even if one fails, so the worker cannot start a + // new refresh while queries from the previous run are still pending. + const results = await Promise.allSettled([ prisma.$queryRaw>` SELECT rows_inserted, rows_updated FROM ohlc.refresh_candles_1m() `, @@ -22,6 +24,10 @@ export async function refreshOhlcAggregates(): Promise { SELECT rows_inserted, rows_updated FROM ohlc.refresh_candles_1d() `, ]); + const [result1m, result1h, result1d] = results.map((result) => { + if (result.status === "rejected") throw new Error("OHLC refresh failed"); + return result.value; + }); const duration = Date.now() - start; @@ -46,26 +52,27 @@ export async function refreshOhlcAggregates(): Promise { : { inserted: 0, updated: 0 }, duration_ms: duration, }; - } catch (err) { - const duration = Date.now() - start; - console.error("[ohlc] Refresh failed:", err); - throw new Error( - `OHLC refresh failed after ${duration}ms: ${(err as Error).message}`, - ); + } catch { + throw new Error("OHLC refresh failed"); } } export function startOhlcRefreshWorker( interval_ms: number = 60_000, ): () => void { + let running = false; const intervalId = setInterval(async () => { + if (running) return; + running = true; try { const result = await refreshOhlcAggregates(); console.log( `[ohlc] Refreshed aggregates (${result.duration_ms}ms): 1m=${result.oneMinute.inserted}, 1h=${result.oneHour.inserted}, 1d=${result.oneDay.inserted}`, ); - } catch (err) { - console.error("[ohlc] Refresh error:", err); + } catch { + console.error("[ohlc] Refresh failed; retrying on the next interval."); + } finally { + running = false; } }, interval_ms); diff --git a/tests/candles-route.test.ts b/tests/candles-route.test.ts new file mode 100644 index 000000000..d7d7e9a75 --- /dev/null +++ b/tests/candles-route.test.ts @@ -0,0 +1,217 @@ +import { inspect } from "node:util"; +import { Prisma } from "@prisma/client"; +import request from "supertest"; + +jest.mock("../src/db", () => ({ + getLastIndexedLedger: jest.fn().mockResolvedValue(1000), + queryTransfers: jest.fn(), + queryAllTransfers: jest.fn(), + queryByTxHash: jest.fn(), + querySummary: jest.fn(), + queryNftTransfers: jest.fn(), + getNftOwner: jest.fn(), + getNftMetadata: jest.fn(), + prisma: { + $queryRawUnsafe: jest.fn(), + $queryRaw: jest.fn(), + }, +})); + +jest.mock("../src/rpc", () => ({ + getLatestLedger: jest.fn().mockResolvedValue(1050), +})); + +jest.mock("../src/indexer", () => ({ + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), + getIndexerStats: jest.fn().mockReturnValue({ uptimeSeconds: 0, totalIndexed: 0 }), +})); + +// The real application still mounts its routers and middleware. Only the +// unrelated GraphQL server is replaced to avoid starting its background work. +jest.mock("../src/graphql/server", () => ({ + createGraphQLMiddleware: () => require("express").Router(), +})); + +import { createApp, clearRpcHealthCache } from "../src/api"; +import { prisma } from "../src/db"; + +const queryCandles = prisma.$queryRawUnsafe as jest.Mock; +const refreshCandles = prisma.$queryRaw as jest.Mock; +const CONTRACT = `C${"A".repeat(55)}`; +const SECRET = "private-database-connection-secret"; + +describe("candles routes through createApp", () => { + const originalEnv = { ...process.env }; + let app: ReturnType; + let errorLog: jest.SpyInstance; + + beforeEach(() => { + process.env.NETWORKS = "testnet"; + process.env.STELLAR_NETWORK = "testnet"; + process.env.CACHE_ENABLED = "false"; + process.env.OHLC_REFRESH_INTERVAL_MS = "60000"; + delete process.env.SKIP_INDEXER; + clearRpcHealthCache(); + queryCandles.mockReset().mockResolvedValue([]); + refreshCandles.mockReset().mockResolvedValue([]); + errorLog = jest.spyOn(console, "error").mockImplementation(() => {}); + app = createApp(); + }); + + afterEach(() => { + errorLog.mockRestore(); + process.env = { ...originalEnv }; + }); + + it.each(["1m", "1h", "1d"])( + "serves the %s aggregate with the default pagination", + async (bucket) => { + queryCandles.mockResolvedValue([ + { + time_bucket: new Date("2026-09-28T12:00:00.000Z"), + contract_id: CONTRACT, + // PostgreSQL NUMERIC values arrive as Prisma.Decimal, so exercise + // real Express JSON serialization without losing decimal precision. + open_price: new Prisma.Decimal("1.100000000000000001"), + high_price: new Prisma.Decimal("1.500000000000000001"), + low_price: new Prisma.Decimal("1.000000000000000001"), + close_price: new Prisma.Decimal("1.300000000000000001"), + volume: new Prisma.Decimal("9007199254740993.0000001"), + tx_count: 7, + }, + ]); + + const res = await request(app).get(`/candles/${bucket}/${CONTRACT}`); + + expect(res.status).toBe(200); + expect(res.body).toEqual({ + bucket, + contractId: CONTRACT, + candles: [ + { + timeBucket: "2026-09-28T12:00:00.000Z", + contractId: CONTRACT, + open: "1.100000000000000001", + high: "1.500000000000000001", + low: "1.000000000000000001", + close: "1.300000000000000001", + volume: "9007199254740993.0000001", + txCount: 7, + }, + ], + }); + expect(queryCandles).toHaveBeenCalledTimes(1); + // Prisma accepts each placeholder value as a separate argument, not an + // array containing all three values as the single value for $1. + expect(queryCandles).toHaveBeenCalledWith( + expect.stringContaining(`FROM ohlc.candles_${bucket}`), + CONTRACT, + 100, + 0, + ); + const sql = queryCandles.mock.calls[0][0] as string; + expect(sql).toMatch(/WHERE contract_id = \$1/); + expect(sql).toMatch(/ORDER BY time_bucket DESC/); + expect(sql).toMatch(/LIMIT \$2 OFFSET \$3/); + expect(sql).not.toContain(CONTRACT); + }, + ); + + it("passes explicit limit and offset as separate SQL parameters", async () => { + const res = await request(app) + .get(`/candles/1h/${CONTRACT}`) + .query({ limit: 25, offset: 50 }); + + expect(res.status).toBe(200); + expect(res.body).toEqual({ bucket: "1h", contractId: CONTRACT, candles: [] }); + expect(queryCandles).toHaveBeenCalledWith(expect.any(String), CONTRACT, 25, 50); + }); + + it("accepts the documented upper page-size boundary", async () => { + const res = await request(app) + .get(`/candles/1d/${CONTRACT}`) + .query({ limit: 1000, offset: 0 }); + + expect(res.status).toBe(200); + expect(queryCandles).toHaveBeenCalledWith(expect.any(String), CONTRACT, 1000, 0); + }); + + it.each([ + ["unsupported bucket", `/candles/5m/${CONTRACT}`], + ["invalid contract ID", "/candles/1m/not-a-contract"], + ["zero limit", `/candles/1m/${CONTRACT}?limit=0`], + ["oversized limit", `/candles/1m/${CONTRACT}?limit=1001`], + ["fractional limit", `/candles/1m/${CONTRACT}?limit=1.5`], + ["non-numeric limit", `/candles/1m/${CONTRACT}?limit=abc`], + ["negative offset", `/candles/1m/${CONTRACT}?offset=-1`], + ["fractional offset", `/candles/1m/${CONTRACT}?offset=1.5`], + ["non-numeric offset", `/candles/1m/${CONTRACT}?offset=abc`], + ])("rejects %s before querying the database", async (_description, url) => { + const res = await request(app).get(url); + + expect(res.status).toBe(400); + expect(res.body.error).toEqual(expect.any(String)); + expect(queryCandles).not.toHaveBeenCalled(); + }); + + it("returns a sanitized 500 when reading candles fails", async () => { + const databaseError = new Error(`Connection failed: ${SECRET}`); + queryCandles.mockRejectedValue(databaseError); + + const res = await request(app).get(`/candles/1m/${CONTRACT}`); + + expect(res.status).toBe(500); + expect(res.body).toEqual({ error: "Failed to query candles" }); + expect(res.text).not.toContain(SECRET); + expect(errorLog.mock.calls.flat()).not.toContain(databaseError); + expect(inspect(errorLog.mock.calls, { depth: null })).not.toContain(SECRET); + }); + + it("does not expose a public refresh endpoint even when candles are enabled", async () => { + const res = await request(app).post("/candles/refresh").send({}); + + expect(res.status).toBe(404); + expect(refreshCandles).not.toHaveBeenCalled(); + expect(queryCandles).not.toHaveBeenCalled(); + }); + + it.each([undefined, "", "0", "-1", "1.5", "60000ms", "2147483648"])( + "leaves candles unmounted for a missing, disabled or invalid interval (%s)", + async (interval) => { + if (interval === undefined) delete process.env.OHLC_REFRESH_INTERVAL_MS; + else process.env.OHLC_REFRESH_INTERVAL_MS = interval; + app = createApp(); + + expect((await request(app).get(`/candles/1m/${CONTRACT}`)).status).toBe(404); + expect(queryCandles).not.toHaveBeenCalled(); + expect(refreshCandles).not.toHaveBeenCalled(); + }, + ); + + it("leaves candles unmounted in API-only mode", async () => { + process.env.SKIP_INDEXER = "true"; + app = createApp(); + + expect((await request(app).get(`/candles/1m/${CONTRACT}`)).status).toBe(404); + expect(queryCandles).not.toHaveBeenCalled(); + }); + + it.each(["testnet,mainnet", "mainnet"])( + "leaves candles unmounted for an incompatible network configuration (%s)", + async (networks) => { + process.env.NETWORKS = networks; + app = createApp(); + + expect((await request(app).get(`/candles/1m/${CONTRACT}`)).status).toBe(404); + expect(queryCandles).not.toHaveBeenCalled(); + }, + ); + + it("rejects a request for a different network before reading aggregates", async () => { + const res = await request(app).get(`/candles/1m/${CONTRACT}?network=mainnet`); + + expect(res.status).toBe(400); + expect(queryCandles).not.toHaveBeenCalled(); + }); +});