feat: yield expiry, bulk epochs endpoint, and OpenTelemetry tracing

backend/package.json

@@ -50,7 +50,11 @@
     "prom-client": "^15.1.3",
     "swagger-ui-express": "^5.0.1",
     "winston": "^3.11.0",
-    "zod": "^3.23.0"
+    "zod": "^3.23.0",
+    "@opentelemetry/sdk-node": "^0.57.0",
+    "@opentelemetry/auto-instrumentations-node": "^0.57.0",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.57.0"
   },
   "devDependencies": {
     "@eslint/js": "^9.15.0",

backend/src/api/controllers/yields.ts

@@ -107,6 +107,35 @@ export async function getYieldSummary(req: Request, res: Response, next: NextFun
   }
 }
 
+export async function getBulkEpochs(req: Request, res: Response, next: NextFunction) {
+  try {
+    const contractId = String(req.params["contractId"]);
+    const from = Number(req.query["from"]);
+    const to = Number(req.query["to"]);
+
+    if (!Number.isInteger(from) || !Number.isInteger(to) || from < 0 || to < 0) {
+      res.status(400).json({ error: "BadRequest", message: "from and to must be non-negative integers" });
+      return;
+    }
+
+    const BULK_EPOCH_LIMIT = 500;
+    if (to - from > BULK_EPOCH_LIMIT) {
+      res.status(400).json({
+        error: "BadRequest",
+        message: `Range exceeds the maximum of ${BULK_EPOCH_LIMIT} epochs`,
+      });
+      return;
+    }
+
+    if (from > to) {
+      res.status(400).json({ error: "BadRequest", message: "from must be less than or equal to to" });
+      return;
+    }
+
+    const epochs = await yieldService.getEpochsBulk(contractId, from, to);
+    res.json(epochs);
+  }
+
 export async function getYieldTimeline(req: Request, res: Response, next: NextFunction) {
   try {
     const contractId = String(req.params["contractId"]);

backend/src/api/routes/yields.ts

@@ -3,6 +3,7 @@ import { z } from "zod";
 import {
   getVaultEpochs,
   getEpochDetail,
+  getBulkEpochs,
   getUserPendingYield,
   getYieldSummary,
   getYieldPerShareHistory,
@@ -56,6 +57,7 @@ export const yieldsRouter = Router();
 yieldsRouter.get("/stream", sseLimitPerIp(), getYieldsStream);
 yieldsRouter.get("/:contractId/summary", getYieldSummary);
 yieldsRouter.get("/:contractId/epochs", validateQuery(epochQuerySchema), getVaultEpochs);
+yieldsRouter.get("/:contractId/epochs/bulk", getBulkEpochs);
 yieldsRouter.get(
   "/:contractId/epochs/:epoch",
   validateParams(epochDetailParamsSchema),

backend/src/config.ts

@@ -125,6 +125,12 @@ const envSchema = z.object({
     .default("15000")
     .transform((v) => parseInt(v, 10))
     .pipe(z.number().int().min(1)),
+  YIELD_CLAIM_EXPIRY_DAYS: z
+    .string()
+    .optional()
+    .transform((v) => (v ? parseInt(v, 10) : null))
+    .pipe(z.number().int().min(1).nullable().default(null)),
+  OTEL_EXPORTER_OTLP_ENDPOINT: z.string().url().optional(),
   SSE_REPLAY_BUFFER: z
     .string()
     .default("100")
@@ -224,6 +230,8 @@ export const config = {
     maxAge: parsed.data.CORS_MAX_AGE,
   },
   sseHeartbeatMs: parsed.data.SSE_HEARTBEAT_MS,
+  yieldClaimExpiryDays: parsed.data.YIELD_CLAIM_EXPIRY_DAYS,
+  otelEndpoint: parsed.data.OTEL_EXPORTER_OTLP_ENDPOINT,
   sseReplayBufferSize: parsed.data.SSE_REPLAY_BUFFER,
   dbPoolAlertWaiting: parsed.data.DB_POOL_ALERT_WAITING,
   rpcErrorRateAlertPct: parsed.data.RPC_ERROR_RATE_ALERT_PCT,

backend/src/db/index.ts

@@ -1,8 +1,11 @@
 import pg from "pg";
 import { performance } from "node:perf_hooks";
+import { trace, SpanStatusCode, context } from "@opentelemetry/api";
 import { config } from "../config.js";
 import { logger } from "../logger.js";
 
+const tracer = trace.getTracer("stellaryield-db");
+
 const { Pool } = pg;
 
 export const pool = new Pool({
@@ -17,26 +20,34 @@ export async function query<T = Record<string, unknown>>(
   sql: string,
   params?: unknown[],
 ): Promise<T[]> {
+  const span = tracer.startSpan("db.query", {}, context.active());
+  span.setAttribute("db.statement", sql.slice(0, 80));
+
   const start = performance.now();
-  const result = await pool.query(sql, params);
-  const durationMs = performance.now() - start;
-  const roundedMs = Math.round(durationMs * 100) / 100;
+  try {
+    const result = await pool.query(sql, params);
+    const durationMs = performance.now() - start;
+    const roundedMs = Math.round(durationMs * 100) / 100;
 
-  if (durationMs > config.db.slowQueryMs) {
-    // Slow queries get higher-visibility logging so they are easy to filter.
-    // The full SQL is included regardless of environment to aid diagnosis (#658).
-    logger.warn(
-      { sql, paramsCount: params?.length ?? 0, durationMs: roundedMs, rowCount: result.rowCount },
-      "slow query",
-    );
-  } else if (logger.level === "debug" || logger.level === "trace") {
-    const firstLine = config.nodeEnv === "production"
-      ? sql.slice(0, 80)
-      : sql;
-    logger.debug({ sql: firstLine, durationMs: roundedMs, rowCount: result.rowCount }, "query");
-  }
+    span.setAttribute("db.response.rows", result.rowCount ?? 0);
 
-  return result.rows;
+    if (durationMs > config.db.slowQueryMs) {
+      logger.warn(
+        { sql, paramsCount: params?.length ?? 0, durationMs: roundedMs, rowCount: result.rowCount },
+        "slow query",
+      );
+    } else if (logger.level === "debug" || logger.level === "trace") {
+      const firstLine = config.nodeEnv === "production" ? sql.slice(0, 80) : sql;
+      logger.debug({ sql: firstLine, durationMs: roundedMs, rowCount: result.rowCount }, "query");
+    }
+
+    return result.rows;
+  } catch (err) {
+    span.setStatus({ code: SpanStatusCode.ERROR, message: String(err) });
+    throw err;
+  } finally {
+    span.end();
+  }
 }
 
 async function validateConnection(): Promise<void> {

backend/src/db/migrations/029_epoch_expiry.sql

@@ -0,0 +1 @@
+ALTER TABLE epochs ADD COLUMN IF NOT EXISTS expires_at TIMESTAMPTZ;

backend/src/index.ts

@@ -1,3 +1,4 @@
+import "./instrumentation.js";
 import { createApp } from "./app.js";
 import { config } from "./config.js";
 import { logger } from "./logger.js";

backend/src/instrumentation.ts

@@ -0,0 +1,22 @@
+import { NodeSDK } from "@opentelemetry/sdk-node";
+import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
+import { SimpleSpanProcessor, NoopSpanProcessor } from "@opentelemetry/sdk-trace-node";
+
+const endpoint = process.env["OTEL_EXPORTER_OTLP_ENDPOINT"];
+
+const spanProcessor = endpoint
+  ? new SimpleSpanProcessor(new OTLPTraceExporter({ url: `${endpoint}/v1/traces` }))
+  : new NoopSpanProcessor();
+
+const sdk = new NodeSDK({
+  serviceName: "stellaryield-backend",
+  spanProcessors: [spanProcessor],
+  instrumentations: [getNodeAutoInstrumentations()],
+});
+
+sdk.start();
+
+process.on("SIGTERM", () => {
+  void sdk.shutdown();
+});

backend/src/services/yield.ts

@@ -1,6 +1,7 @@
 import type { Epoch } from "../types/index.js";
 import { query } from "../db/index.js";
 import { cacheGet, cacheSet, cacheDel } from "../cache/redis.js";
+import { config } from "../config.js";
 
 const EPOCHS_CACHE_TTL = 30;
 const PENDING_YIELD_CACHE_TTL = 10;
@@ -301,11 +302,13 @@ export class YieldService {
       epoch: number;
       yield_amount: string;
       total_shares: string;
+      expires_at: Date | null;
     }>(
-      `SELECT e.epoch, e.yield_amount, e.total_shares
+      `SELECT e.epoch, e.yield_amount, e.total_shares, e.expires_at
        FROM epochs e
        JOIN vaults v ON e.vault_id = v.id
        WHERE v.contract_id = $1
+         AND (e.expires_at IS NULL OR e.expires_at > NOW())
        ORDER BY e.epoch ASC`,
       [contractId],
     );
@@ -408,15 +411,48 @@ export class YieldService {
     yieldAmount: string,
     totalShares: string,
   ): Promise<void> {
+    const expiryDays = config.yieldClaimExpiryDays;
+    const expiresAt = expiryDays
+      ? new Date(Date.now() + expiryDays * 24 * 60 * 60 * 1000)
+      : null;
+
     await query(
-      `INSERT INTO epochs (vault_id, epoch, yield_amount, total_shares, distributed_at)
-       VALUES ($1, $2, $3, $4, NOW())
+      `INSERT INTO epochs (vault_id, epoch, yield_amount, total_shares, distributed_at, expires_at)
+       VALUES ($1, $2, $3, $4, NOW(), $5)
        ON CONFLICT (vault_id, epoch) DO NOTHING`,
-      [vaultId, epoch, yieldAmount, totalShares],
+      [vaultId, epoch, yieldAmount, totalShares, expiresAt],
     );
     await cacheDel(`epochs:*`);
   }
 
+  async getEpochsBulk(
+    contractId: string,
+    from: number,
+    to: number,
+  ): Promise<Array<{ epoch: number; yieldAmount: string; totalShares: string; yieldPerShare: string; distributedAt: string | null }>> {
+    const rows = await query<{
+      epoch: number;
+      yield_amount: string;
+      total_shares: string;
+      distributed_at: Date | null;
+    }>(
+      `SELECT e.epoch, e.yield_amount, e.total_shares, e.distributed_at
+       FROM epochs e
+       JOIN vaults v ON e.vault_id = v.id
+       WHERE v.contract_id = $1 AND e.epoch >= $2 AND e.epoch <= $3
+       ORDER BY e.epoch ASC`,
+      [contractId, from, to],
+    );
+
+    return rows.map((row) => ({
+      epoch: row.epoch,
+      yieldAmount: row.yield_amount,
+      totalShares: row.total_shares,
+      yieldPerShare: this.formatYieldPerShare(row.yield_amount, row.total_shares),
+      distributedAt: row.distributed_at ? row.distributed_at.toISOString() : null,
+    }));
+  }
+
   async getYieldPerShareHistory(
     contractId: string,
     from?: Date,