fix(api): gate candles routes with the refresh worker

.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:

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)                         |

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:-}

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<string, unknown> {
+    let paths: Record<string, unknown> = {};
+    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}");
+    },
+  );
+});

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<Array<{ rows_inserted: number; rows_updated: number }>>((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();
+  });
+});

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<string | symbol, () => Promise<void>>();
+
+  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<void>);
+      return process;
+    });
+  });
+
+  afterEach(() => {
+    process.env = { ...originalEnv };
+    jest.restoreAllMocks();
+  });
+
+  async function startService(): Promise<void> {
+    jest.isolateModules(() => {
+      require("../index");
+    });
+    await new Promise<void>((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);
+  });
+});

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());