diff --git a/backend/package.json b/backend/package.json index 1cbaa24..c2c0a42 100644 --- a/backend/package.json +++ b/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", diff --git a/backend/src/api/controllers/yields.ts b/backend/src/api/controllers/yields.ts index 204a620..f62aa26 100644 --- a/backend/src/api/controllers/yields.ts +++ b/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"]); diff --git a/backend/src/api/routes/yields.ts b/backend/src/api/routes/yields.ts index a900df6..937de0d 100644 --- a/backend/src/api/routes/yields.ts +++ b/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), diff --git a/backend/src/config.ts b/backend/src/config.ts index 8654e41..0c8242a 100644 --- a/backend/src/config.ts +++ b/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, diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index 1a92b0d..e6361ca 100644 --- a/backend/src/db/index.ts +++ b/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>( sql: string, params?: unknown[], ): Promise { + 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 { diff --git a/backend/src/db/migrations/029_epoch_expiry.sql b/backend/src/db/migrations/029_epoch_expiry.sql new file mode 100644 index 0000000..e9a5742 --- /dev/null +++ b/backend/src/db/migrations/029_epoch_expiry.sql @@ -0,0 +1 @@ +ALTER TABLE epochs ADD COLUMN IF NOT EXISTS expires_at TIMESTAMPTZ; diff --git a/backend/src/index.ts b/backend/src/index.ts index e9e3dc7..9e33ffe 100644 --- a/backend/src/index.ts +++ b/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"; diff --git a/backend/src/instrumentation.ts b/backend/src/instrumentation.ts new file mode 100644 index 0000000..371a36d --- /dev/null +++ b/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(); +}); diff --git a/backend/src/services/yield.ts b/backend/src/services/yield.ts index 140729b..bc8c985 100644 --- a/backend/src/services/yield.ts +++ b/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 { + 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> { + 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,