From 07bab35f82f53a60105841a6f8a64f1a25575701 Mon Sep 17 00:00:00 2001 From: wheval Date: Fri, 28 Aug 2026 10:20:07 +0100 Subject: [PATCH 1/4] feat: adaptive per-wallet query cost governor with rolling Redis budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the query cost governor from #755: assigns a cost unit to every request, tracks a rolling per-caller spend in Redis, and rejects requests that would exceed the budget with 429 query_budget_exceeded. Identity: most of the routes this protects (creator list, holders, search) are public reads with no wallet-auth middleware in front of them today, unlike the mutating routes requireStellarSignature() already covers. The governor resolves the caller's wallet from a JWT if one is present (the same one requireJwtAuth checks) without rejecting when it's absent or invalid — falling back to an IP-keyed budget so the governor actually protects the public routes the issue names, not only already-authenticated ones. - src/utils/query-cost.utils.ts: route-pattern -> cost matching (works off req.path since the governor runs ahead of route resolution, before req.route is populated) and cost computation with the limit-param multiplier. - src/constants/query-cost.constants.ts: default cost map. The issue's own example routes (GET /search, GET /creators/:id/history) don't exist verbatim in this codebase; substituted the closest real equivalents (GET /keys/search, GET /creators/:id/stats). - src/middlewares/query-cost-governor.middleware.ts: the governor itself. Evicts expired entries, sums the caller's remaining cost, and only admits the request if it fits the budget -- true check-before-append, not append-then-check. Mounted once, globally, in src/modules/index.ts (ahead of every route group) rather than threaded into each route file individually. Same fail-open-on-Redis-error posture as the existing wallet-rate-limit.middleware.ts, and the same non-atomic evict/read-then-write tradeoff (documented inline) rather than a Lua script, matching this codebase's existing accepted rigor level for Redis rate limiting. - src/modules/admin/query-cost.controllers.ts: POST /internal/qcost/reset/:walletAddress, added to the existing sequencer router (already mounted at /internal with no app-level auth -- network isolation only, same convention as the existing /internal/sequencer/clear-drift/:creatorWallet). - QUERY_COST_BUDGET / QUERY_COST_WINDOW_MS / QUERY_COST_MAP_JSON / QUERY_COST_ADMIN_WALLETS added to config.schema.ts, all optional/defaulted. 26 new tests across the three new files, all passing. Verified no new TypeScript errors via a full-repo `tsc --noEmit` diff against a clean checkout (identical 78 pre-existing errors before and after). `pnpm build` currently fails on main itself for many unrelated pre-existing reasons (server.ts importing a nonexistent connectRedis export, Prisma schema drift in ownership/wallet-following, undefined handlers in wallets.routes.ts and keys.routes.ts, more missing config schema fields) -- confirmed via the same stash-diff technique, none of it in files this PR touches. Closes accesslayerorg/accesslayer-server#755 --- .env.example | 6 + src/config.schema.ts | 18 ++ src/constants/query-cost.constants.ts | 29 ++ .../query-cost-governor.middleware.test.ts | 285 ++++++++++++++++++ .../query-cost-governor.middleware.ts | 223 ++++++++++++++ .../admin/query-cost.controllers.test.ts | 54 ++++ src/modules/admin/query-cost.controllers.ts | 41 +++ src/modules/admin/sequencer.routes.ts | 3 + src/modules/index.ts | 9 + src/utils/query-cost.utils.test.ts | 83 +++++ src/utils/query-cost.utils.ts | 140 +++++++++ 11 files changed, 891 insertions(+) create mode 100644 src/constants/query-cost.constants.ts create mode 100644 src/middlewares/query-cost-governor.middleware.test.ts create mode 100644 src/middlewares/query-cost-governor.middleware.ts create mode 100644 src/modules/admin/query-cost.controllers.test.ts create mode 100644 src/modules/admin/query-cost.controllers.ts create mode 100644 src/utils/query-cost.utils.test.ts create mode 100644 src/utils/query-cost.utils.ts diff --git a/.env.example b/.env.example index 1972614..0ba3fb1 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,12 @@ JWT_ACCESS_TOKEN_TTL_SECONDS=900 REDIS_URL=redis://localhost:6379 ENABLE_REDIS_CACHE=true +# Query cost governor (#755): rolling per-wallet/per-IP database query budget +QUERY_COST_BUDGET=200 +QUERY_COST_WINDOW_MS=60000 +# QUERY_COST_MAP_JSON={"GET /search": 8} +# QUERY_COST_ADMIN_WALLETS= + # Key trade lockup (seconds between a buy and when the keys unlock for sale) LOCKUP_DURATION_SECONDS=0 diff --git a/src/config.schema.ts b/src/config.schema.ts index 433f022..09daeed 100644 --- a/src/config.schema.ts +++ b/src/config.schema.ts @@ -222,6 +222,24 @@ export const envSchema = z // Left unset by default, so no caller is trusted unless configured. TRACE_ID_TRUSTED_TOKEN: optionalNonEmptyString, INTERNAL_SERVICE_KEY: optionalNonEmptyString, + + // Query cost governor (#755): rolling per-wallet (or per-IP, when + // unauthenticated) database query budget. See + // src/middlewares/query-cost-governor.middleware.ts. + QUERY_COST_BUDGET: z.coerce.number().int().positive().default(200), + QUERY_COST_WINDOW_MS: z.coerce + .number() + .int() + .positive() + .default(60_000), + // JSON object overriding/extending the default route->cost map in + // src/constants/query-cost.constants.ts, e.g. + // '{"GET /search": 8, "GET /custom-route": 2}'. Merged over the + // defaults, not a full replacement, so operators only need to + // specify what differs. + QUERY_COST_MAP_JSON: optionalNonEmptyString, + // Comma-separated wallet addresses that bypass the governor entirely. + QUERY_COST_ADMIN_WALLETS: optionalNonEmptyString, HORIZON_WEBHOOK_SECRET: optionalNonEmptyString, WEBHOOK_RETRY_BASE_DELAY_MS: z.coerce .number() diff --git a/src/constants/query-cost.constants.ts b/src/constants/query-cost.constants.ts new file mode 100644 index 0000000..19deec2 --- /dev/null +++ b/src/constants/query-cost.constants.ts @@ -0,0 +1,29 @@ +// src/constants/query-cost.constants.ts +// Default route->cost map for the query cost governor (#755). +// +// Keys are `${METHOD} ${pattern}`, where `pattern` uses Express-style +// `:param` segments (matched via src/utils/query-cost.utils.ts, not +// req.route — the governor runs as router-level middleware, before Express +// resolves the specific route, so req.route isn't populated yet). +// +// The issue's own example routes (`GET /search`, `GET /creators/:id/history`) +// don't exist verbatim in this codebase; substituted for the closest real +// equivalents (`GET /keys/search`, `GET /creators/:id/stats`) — see the PR +// description for the full mapping rationale. +export const DEFAULT_QUERY_COST_MAP: Record = { + 'GET /creators': 1, + 'GET /creators/:id/holders': 3, + 'GET /creators/:id/stats': 2, + 'GET /keys/search': 5, +}; + +/** Cost applied to any authenticated route with no explicit entry in the map. */ +export const DEFAULT_QUERY_COST = 1; + +/** + * Route patterns exempt from query cost governance entirely: health checks + * (must stay reachable regardless of load) and the governor's own internal + * management routes (resetting a wallet's budget must never itself be + * throttled by that same budget). + */ +export const QUERY_COST_EXEMPT_PATH_PREFIXES = ['/health', '/internal/qcost']; diff --git a/src/middlewares/query-cost-governor.middleware.test.ts b/src/middlewares/query-cost-governor.middleware.test.ts new file mode 100644 index 0000000..3baf113 --- /dev/null +++ b/src/middlewares/query-cost-governor.middleware.test.ts @@ -0,0 +1,285 @@ +// Unit tests for the adaptive query cost governor (#755). + +const mockEnvConfig: { + INTERNAL_SERVICE_KEY?: string; + QUERY_COST_BUDGET: number; + QUERY_COST_WINDOW_MS: number; + QUERY_COST_MAP_JSON?: string; + QUERY_COST_ADMIN_WALLETS?: string; +} = { + INTERNAL_SERVICE_KEY: undefined, + QUERY_COST_BUDGET: 200, + QUERY_COST_WINDOW_MS: 60_000, + QUERY_COST_MAP_JSON: undefined, + QUERY_COST_ADMIN_WALLETS: undefined, +}; + +jest.mock('../config', () => ({ + envConfig: mockEnvConfig, +})); + +jest.mock('../utils/logger.utils', () => ({ + logger: { warn: jest.fn(), error: jest.fn(), debug: jest.fn(), info: jest.fn() }, +})); + +function buildFakeRedis() { + const store = new Map>(); + + return { + zremrangebyscore: jest.fn(async (key: string, _min: number, max: number) => { + const entries = store.get(key) ?? []; + store.set( + key, + entries.filter(entry => entry.score > max) + ); + }), + zrange: jest.fn(async (key: string, _start: number, _stop: number, withScores?: string) => { + const entries = (store.get(key) ?? []).sort((a, b) => a.score - b.score); + if (withScores === 'WITHSCORES') { + const first = entries[0]; + return first ? [first.member, String(first.score)] : []; + } + return entries.map(entry => entry.member); + }), + zadd: jest.fn(async (key: string, score: number, member: string) => { + const entries = store.get(key) ?? []; + entries.push({ score, member }); + store.set(key, entries); + }), + pexpire: jest.fn(async () => 1), + del: jest.fn(async (key: string) => { + store.delete(key); + }), + __store: store, + }; +} + +jest.mock('../utils/redis.utils', () => ({ + getRedis: jest.fn(), +})); + +const mockVerifyWalletAccessToken = jest.fn(); +jest.mock('../utils/jwt.utils', () => ({ + extractBearerToken: (header: unknown) => + typeof header === 'string' && header.startsWith('Bearer ') + ? header.slice(7) + : undefined, + verifyWalletAccessToken: (token: string) => mockVerifyWalletAccessToken(token), +})); + +import { getRedis } from '../utils/redis.utils'; +import { queryCostGovernor } from './query-cost-governor.middleware'; + +const mockGetRedis = getRedis as jest.Mock; + +function makeReq(overrides: Partial> = {}): any { + return { + method: 'GET', + path: '/creators', + query: {}, + headers: {}, + ip: '203.0.113.5', + ...overrides, + }; +} + +function makeRes(): any { + const res: any = {}; + res.status = jest.fn().mockReturnValue(res); + res.json = jest.fn().mockReturnValue(res); + res.set = jest.fn().mockReturnValue(res); + return res; +} + +describe('queryCostGovernor', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockEnvConfig.INTERNAL_SERVICE_KEY = undefined; + mockEnvConfig.QUERY_COST_BUDGET = 200; + mockEnvConfig.QUERY_COST_WINDOW_MS = 60_000; + mockEnvConfig.QUERY_COST_MAP_JSON = undefined; + mockEnvConfig.QUERY_COST_ADMIN_WALLETS = undefined; + mockVerifyWalletAccessToken.mockReset(); + }); + + it('admits requests summing to exactly the budget, keyed per IP when unauthenticated', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + mockEnvConfig.QUERY_COST_BUDGET = 10; + const governor = queryCostGovernor(); + + // 10 requests at cost 1 (GET /creators) = exactly the budget. + for (let i = 0; i < 10; i++) { + const req = makeReq(); + const res = makeRes(); + const next = jest.fn(); + await governor(req, res, next); + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + } + }); + + it('throttles the request that pushes total cost over budget, with Retry-After and reset headers', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + mockEnvConfig.QUERY_COST_BUDGET = 5; + const governor = queryCostGovernor(); + + // GET /creators/:id/holders costs 3 by default. + for (let i = 0; i < 1; i++) { + const req = makeReq({ path: '/creators/abc/holders' }); + const res = makeRes(); + await governor(req, res, jest.fn()); + } + + const req = makeReq({ path: '/creators/abc/holders' }); + const res = makeRes(); + const next = jest.fn(); + await governor(req, res, next); + + expect(next).not.toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(429); + expect(res.set).toHaveBeenCalledWith('Retry-After', expect.any(String)); + expect(res.set).toHaveBeenCalledWith( + 'X-Query-Budget-Reset', + expect.any(String) + ); + const body = res.json.mock.calls[0][0]; + expect(body.type).toBe('query_budget_exceeded'); + }); + + it('sets X-Query-Cost and X-Query-Budget-Remaining on successful responses', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + const governor = queryCostGovernor(); + + const req = makeReq({ path: '/creators/abc/holders' }); + const res = makeRes(); + await governor(req, res, jest.fn()); + + expect(res.set).toHaveBeenCalledWith('X-Query-Cost', '3'); + expect(res.set).toHaveBeenCalledWith( + 'X-Query-Budget-Remaining', + String(mockEnvConfig.QUERY_COST_BUDGET - 3) + ); + }); + + it('multiplies cost by the limit query param', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + const governor = queryCostGovernor(); + + const req = makeReq({ path: '/creators/abc/holders', query: { limit: '10' } }); + const res = makeRes(); + await governor(req, res, jest.fn()); + + expect(res.set).toHaveBeenCalledWith('X-Query-Cost', '30'); + }); + + it('allows requests again after the rolling window expires', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + mockEnvConfig.QUERY_COST_BUDGET = 1; + mockEnvConfig.QUERY_COST_WINDOW_MS = 50; + const governor = queryCostGovernor(); + + const first = makeReq(); + await governor(first, makeRes(), jest.fn()); + + const blocked = makeReq(); + const blockedRes = makeRes(); + const blockedNext = jest.fn(); + await governor(blocked, blockedRes, blockedNext); + expect(blockedNext).not.toHaveBeenCalled(); + + await new Promise(resolve => setTimeout(resolve, 60)); + + const afterWindow = makeReq(); + const afterRes = makeRes(); + const afterNext = jest.fn(); + await governor(afterWindow, afterRes, afterNext); + expect(afterNext).toHaveBeenCalledTimes(1); + }); + + it('bypasses the governor entirely for admin wallets', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + mockEnvConfig.QUERY_COST_BUDGET = 1; + mockEnvConfig.QUERY_COST_ADMIN_WALLETS = 'GADMIN123, GOTHER456'; + mockVerifyWalletAccessToken.mockReturnValue({ wallet: 'GADMIN123' }); + const governor = queryCostGovernor(); + + for (let i = 0; i < 5; i++) { + const req = makeReq({ + path: '/creators/abc/holders', + headers: { authorization: 'Bearer admin-token' }, + }); + const res = makeRes(); + const next = jest.fn(); + await governor(req, res, next); + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + } + }); + + it('bypasses the governor for internal service calls', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + mockEnvConfig.QUERY_COST_BUDGET = 0; + mockEnvConfig.INTERNAL_SERVICE_KEY = 'internal-secret'; + const governor = queryCostGovernor(); + + const req = makeReq({ headers: { 'x-internal-service-key': 'internal-secret' } }); + const res = makeRes(); + const next = jest.fn(); + await governor(req, res, next); + + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + }); + + it('exempts /health and its own /internal/qcost routes', async () => { + const redis = buildFakeRedis(); + mockGetRedis.mockReturnValue(redis); + mockEnvConfig.QUERY_COST_BUDGET = 0; + const governor = queryCostGovernor(); + + for (const path of ['/health', '/internal/qcost/reset/GABC']) { + const req = makeReq({ path, method: 'POST' }); + const res = makeRes(); + const next = jest.fn(); + await governor(req, res, next); + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + } + }); + + it('fails open when Redis is unavailable', async () => { + mockGetRedis.mockReturnValue(null); + mockEnvConfig.QUERY_COST_BUDGET = 0; + const governor = queryCostGovernor(); + + const req = makeReq(); + const res = makeRes(); + const next = jest.fn(); + await governor(req, res, next); + + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + }); + + it('fails open when a Redis command throws', async () => { + const redis = buildFakeRedis(); + redis.zremrangebyscore.mockRejectedValueOnce(new Error('connection reset')); + mockGetRedis.mockReturnValue(redis); + const governor = queryCostGovernor(); + + const req = makeReq(); + const res = makeRes(); + const next = jest.fn(); + await governor(req, res, next); + + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + }); +}); diff --git a/src/middlewares/query-cost-governor.middleware.ts b/src/middlewares/query-cost-governor.middleware.ts new file mode 100644 index 0000000..0cec4eb --- /dev/null +++ b/src/middlewares/query-cost-governor.middleware.ts @@ -0,0 +1,223 @@ +// src/middlewares/query-cost-governor.middleware.ts +// Adaptive per-wallet (or per-IP, when unauthenticated) database query cost +// governor (#755). +// +// A single wallet firing expensive paginated/search/analytics queries can +// saturate the connection pool for everyone. This assigns a cost unit to +// every request (see src/utils/query-cost.utils.ts), tracks a rolling sum +// of costs per caller in a Redis sorted set, and rejects requests that would +// push the caller over budget with 429 query_budget_exceeded. +// +// Identity: most of the routes this is meant to protect (creator list, +// holders, search) are public reads with no wallet-auth middleware in front +// of them today, so "per-wallet" only applies when the caller sent a valid +// JWT (the same one requireJwtAuth checks) — decoded here without rejecting +// when absent/invalid, unlike requireJwtAuth. Anonymous callers still get a +// real budget, keyed on IP, so the governor actually protects the public +// routes named in the issue rather than only the already-authenticated ones. +// +// Concurrency note: like wallet-rate-limit.middleware.ts, this evicts/reads +// then conditionally writes in separate round trips rather than a single +// atomic Lua script — under a burst of truly concurrent requests from the +// same caller there's a small window where more than the budget could be +// admitted. Matches this codebase's existing accepted tradeoff for Redis +// rate limiting rather than introducing a different rigor level for this +// one feature. + +import type { Request, Response, NextFunction } from 'express'; +import { randomUUID } from 'crypto'; +import { getRedis } from '../utils/redis.utils'; +import { envConfig } from '../config'; +import { logger } from '../utils/logger.utils'; +import { extractBearerToken, verifyWalletAccessToken } from '../utils/jwt.utils'; +import { + buildQueryCostRedisKey, + compileCostMap, + computeQueryCost, + matchCostRoute, + type CompiledCostRoute, +} from '../utils/query-cost.utils'; +import { QUERY_COST_EXEMPT_PATH_PREFIXES } from '../constants/query-cost.constants'; + +const INTERNAL_SERVICE_HEADER = 'x-internal-service-key'; + +export interface QueryCostRequest extends Request { + walletAddress?: string; + queryCost?: { cost: number; identity: string }; +} + +function isInternalServiceCall(req: Request): boolean { + if (!envConfig.INTERNAL_SERVICE_KEY) return false; + const provided = req.headers[INTERNAL_SERVICE_HEADER]; + const value = Array.isArray(provided) ? provided[0] : provided; + return value === envConfig.INTERNAL_SERVICE_KEY; +} + +function isExemptPath(path: string): boolean { + return QUERY_COST_EXEMPT_PATH_PREFIXES.some(prefix => + path.startsWith(prefix) + ); +} + +/** Best-effort wallet resolution: never rejects on a missing/invalid token. */ +function resolveWalletAddress(req: Request): string | undefined { + const token = extractBearerToken(req.headers.authorization); + if (!token) return undefined; + try { + return verifyWalletAccessToken(token).wallet; + } catch { + return undefined; + } +} + +function parseAdminWallets(raw: string | undefined): Set { + if (!raw) return new Set(); + return new Set( + raw + .split(',') + .map(wallet => wallet.trim().toLowerCase()) + .filter(Boolean) + ); +} + +/** Encodes a sorted-set member as ":" so cost survives eviction/summation without a second data structure. */ +function encodeMember(cost: number): string { + return `${cost}:${randomUUID()}`; +} + +function decodeCost(member: string): number { + const separator = member.indexOf(':'); + const cost = Number.parseInt( + separator === -1 ? member : member.slice(0, separator), + 10 + ); + return Number.isFinite(cost) ? cost : 0; +} + +export interface QueryCostGovernorOptions { + /** Rolling window in milliseconds. Defaults to envConfig.QUERY_COST_WINDOW_MS. */ + windowMs?: number; + /** Budget per window. Defaults to envConfig.QUERY_COST_BUDGET. */ + budget?: number; +} + +export function queryCostGovernor(options: QueryCostGovernorOptions = {}) { + const windowMs = options.windowMs ?? envConfig.QUERY_COST_WINDOW_MS; + const budget = options.budget ?? envConfig.QUERY_COST_BUDGET; + const adminWallets = parseAdminWallets(envConfig.QUERY_COST_ADMIN_WALLETS); + + let compiledRoutes: CompiledCostRoute[]; + try { + compiledRoutes = compileCostMap(envConfig.QUERY_COST_MAP_JSON); + } catch (error) { + logger.error( + { type: 'query_cost_config_invalid', error }, + 'Invalid QUERY_COST_MAP_JSON; falling back to defaults' + ); + compiledRoutes = compileCostMap(undefined); + } + + return async ( + req: QueryCostRequest, + res: Response, + next: NextFunction + ): Promise => { + if (isExemptPath(req.path) || isInternalServiceCall(req)) { + next(); + return; + } + + const walletAddress = resolveWalletAddress(req); + req.walletAddress = walletAddress; + + if (walletAddress && adminWallets.has(walletAddress.toLowerCase())) { + next(); + return; + } + + const identity = walletAddress + ? `wallet:${walletAddress}` + : `ip:${req.ip ?? 'unknown'}`; + + const matched = matchCostRoute(compiledRoutes, req.method, req.path); + const cost = computeQueryCost(matched, req.query.limit); + + const redis = getRedis(); + if (!redis) { + // Fail open, same as wallet-rate-limit.middleware.ts: caching/rate + // infra being down must never take the API down. + req.queryCost = { cost, identity }; + res.set('X-Query-Cost', String(cost)); + next(); + return; + } + + const key = buildQueryCostRedisKey(identity); + const now = Date.now(); + const windowStart = now - windowMs; + + try { + await redis.zremrangebyscore(key, 0, windowStart); + const members = await redis.zrange(key, 0, '-1'); + const spent = members.reduce( + (sum, member) => sum + decodeCost(member), + 0 + ); + + if (spent + cost > budget) { + const oldest = await redis.zrange(key, 0, '0', 'WITHSCORES'); + const oldestTimestamp = oldest[1] ? Number(oldest[1]) : now; + const resetAtMs = oldestTimestamp + windowMs; + const retryAfterSeconds = Math.max( + 1, + Math.ceil((resetAtMs - now) / 1000) + ); + + logger.warn( + { + type: 'query_budget_exceeded', + identity, + route: req.path, + method: req.method, + cost, + spent, + budget, + }, + 'Query budget exceeded' + ); + + res + .status(429) + .set('Retry-After', String(retryAfterSeconds)) + .set('X-Query-Cost', String(cost)) + .set('X-Query-Budget-Remaining', String(Math.max(0, budget - spent))) + .set('X-Query-Budget-Reset', String(Math.floor(resetAtMs / 1000))) + .json({ + type: 'query_budget_exceeded', + message: 'Query budget exceeded for this window.', + retryAfterSeconds, + timestamp: new Date().toISOString(), + }); + return; + } + + await redis.zadd(key, now, encodeMember(cost)); + await redis.pexpire(key, windowMs); + + req.queryCost = { cost, identity }; + res.set('X-Query-Cost', String(cost)); + res.set( + 'X-Query-Budget-Remaining', + String(Math.max(0, budget - spent - cost)) + ); + next(); + } catch (error) { + logger.error( + { error, identity, route: req.path }, + 'Query cost governor check failed; allowing request through (fail open)' + ); + res.set('X-Query-Cost', String(cost)); + next(); + } + }; +} diff --git a/src/modules/admin/query-cost.controllers.test.ts b/src/modules/admin/query-cost.controllers.test.ts new file mode 100644 index 0000000..35bcef7 --- /dev/null +++ b/src/modules/admin/query-cost.controllers.test.ts @@ -0,0 +1,54 @@ +import { httpResetQueryCost } from './query-cost.controllers'; + +const mockDel = jest.fn(); + +jest.mock('../../utils/redis.utils', () => ({ + getRedis: jest.fn(() => ({ del: mockDel })), +})); + +jest.mock('../../utils/logger.utils', () => ({ + logger: { warn: jest.fn(), error: jest.fn(), debug: jest.fn(), info: jest.fn() }, +})); + +describe('httpResetQueryCost', () => { + const next = jest.fn(); + + function createRes(): any { + const res: any = {}; + res.status = jest.fn().mockReturnValue(res); + res.json = jest.fn().mockReturnValue(res); + return res; + } + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('clears the wallet-scoped Redis key and returns success', async () => { + const req: any = { params: { walletAddress: 'GABC123' } }; + const res = createRes(); + + await httpResetQueryCost(req, res, next); + + expect(mockDel).toHaveBeenCalledWith('qcost:wallet:GABC123'); + expect(res.json).toHaveBeenCalledWith( + expect.objectContaining({ + success: true, + data: expect.objectContaining({ + walletAddress: 'GABC123', + status: 'reset', + }), + }) + ); + }); + + it('rejects a missing walletAddress param', async () => { + const req: any = { params: {} }; + const res = createRes(); + + await httpResetQueryCost(req, res, next); + + expect(mockDel).not.toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(400); + }); +}); diff --git a/src/modules/admin/query-cost.controllers.ts b/src/modules/admin/query-cost.controllers.ts new file mode 100644 index 0000000..bb0eb53 --- /dev/null +++ b/src/modules/admin/query-cost.controllers.ts @@ -0,0 +1,41 @@ +import { AsyncController } from '../../types/auth.types'; +import { getRedis } from '../../utils/redis.utils'; +import { buildQueryCostRedisKey } from '../../utils/query-cost.utils'; +import { sendSuccess, sendValidationError } from '../../utils/api-response.utils'; +import { logger } from '../../utils/logger.utils'; + +/** + * POST /internal/qcost/reset/:walletAddress + * + * Clears a wallet's rolling query-cost budget immediately. Internal-network + * route (see src/modules/index.ts's mount and README) — same convention as + * the existing /internal/sequencer/clear-drift/:creatorWallet. + */ +export const httpResetQueryCost: AsyncController = async (req, res, next) => { + try { + const rawParam = req.params.walletAddress; + const walletAddress = Array.isArray(rawParam) ? rawParam[0] : rawParam; + if (!walletAddress) { + sendValidationError(res, 'Missing walletAddress parameter'); + return; + } + + const redis = getRedis(); + if (redis) { + await redis.del(buildQueryCostRedisKey(`wallet:${walletAddress}`)); + } + + logger.warn( + { type: 'query_cost_reset', walletAddress }, + 'Query cost budget reset by operator' + ); + + sendSuccess(res, { + walletAddress, + status: 'reset', + message: `Query cost budget cleared for ${walletAddress}`, + }); + } catch (err) { + next(err); + } +}; diff --git a/src/modules/admin/sequencer.routes.ts b/src/modules/admin/sequencer.routes.ts index fa6c605..81d2530 100644 --- a/src/modules/admin/sequencer.routes.ts +++ b/src/modules/admin/sequencer.routes.ts @@ -1,8 +1,11 @@ import { Router } from 'express'; import { httpClearDrift } from './sequencer.controllers'; +import { httpResetQueryCost } from './query-cost.controllers'; const sequencerRouter = Router(); sequencerRouter.post('/sequencer/clear-drift/:creatorWallet', httpClearDrift); +// Query cost governor admin override (#755). +sequencerRouter.post('/qcost/reset/:walletAddress', httpResetQueryCost); export default sequencerRouter; diff --git a/src/modules/index.ts b/src/modules/index.ts index 105cd6e..27592b8 100644 --- a/src/modules/index.ts +++ b/src/modules/index.ts @@ -1,4 +1,5 @@ import { routeBodySizeLimit } from '../middlewares/body-size-limit.middleware'; +import { queryCostGovernor } from '../middlewares/query-cost-governor.middleware'; import { Router } from 'express'; import authRouter from './auth/auth.routes'; import healthRouter from './health/health.routes'; @@ -25,6 +26,14 @@ import { BASE as CREATORS_BASE } from '../constants/creator.constants'; const router = Router(); +// Adaptive per-wallet/per-IP database query cost governor (#755). Mounted +// ahead of route resolution (so it matches on req.path, not req.route — see +// query-cost.utils.ts) and ahead of every group below, so it covers the +// whole API surface rather than needing to be threaded into each route +// individually. Exempts /health and its own /internal/qcost management +// routes (see QUERY_COST_EXEMPT_PATH_PREFIXES). +router.use(queryCostGovernor()); + // Each group gets its own JSON body parser so its size limit can be tuned // independently via BODY_SIZE_LIMIT_ env vars (see // docs/body-size-limits.md). Groups without a dedicated override share diff --git a/src/utils/query-cost.utils.test.ts b/src/utils/query-cost.utils.test.ts new file mode 100644 index 0000000..be60e91 --- /dev/null +++ b/src/utils/query-cost.utils.test.ts @@ -0,0 +1,83 @@ +import { + compileCostMap, + computeQueryCost, + matchCostRoute, +} from './query-cost.utils'; + +describe('compileCostMap', () => { + it('compiles the default map and matches param segments', () => { + const routes = compileCostMap(); + const matched = matchCostRoute(routes, 'GET', '/creators/abc-123/holders'); + expect(matched?.baseCost).toBe(3); + }); + + it('does not match a different method for the same path', () => { + const routes = compileCostMap(); + expect(matchCostRoute(routes, 'POST', '/creators')).toBeNull(); + }); + + it('merges a JSON override over the defaults without dropping unrelated entries', () => { + const routes = compileCostMap('{"GET /custom": 9}'); + expect(matchCostRoute(routes, 'GET', '/custom')?.baseCost).toBe(9); + expect(matchCostRoute(routes, 'GET', '/creators')?.baseCost).toBe(1); + }); + + it('lets a JSON override replace a default entry', () => { + const routes = compileCostMap('{"GET /creators": 4}'); + expect(matchCostRoute(routes, 'GET', '/creators')?.baseCost).toBe(4); + }); + + it('rejects malformed JSON', () => { + expect(() => compileCostMap('{not json')).toThrow(/valid JSON/); + }); + + it('rejects a non-object JSON value', () => { + expect(() => compileCostMap('[1,2,3]')).toThrow(/JSON object/); + }); + + it('rejects a non-positive-number cost', () => { + expect(() => compileCostMap('{"GET /x": -1}')).toThrow(/positive number/); + expect(() => compileCostMap('{"GET /x": "5"}')).toThrow(/positive number/); + }); + + it('rejects a key with no method prefix', () => { + expect(() => compileCostMap('{"/no-method": 1}')).toThrow(/METHOD \/pattern/); + }); +}); + +describe('computeQueryCost', () => { + it('uses the default cost when no route matched', () => { + expect(computeQueryCost(null, undefined)).toBe(1); + }); + + it('uses the matched route base cost with no limit param', () => { + const routes = compileCostMap(); + const matched = matchCostRoute(routes, 'GET', '/creators/abc/holders'); + expect(computeQueryCost(matched, undefined)).toBe(3); + }); + + it('multiplies by a numeric limit param', () => { + const routes = compileCostMap(); + const matched = matchCostRoute(routes, 'GET', '/creators/abc/holders'); + expect(computeQueryCost(matched, '100')).toBe(300); + }); + + it('ignores a limit of 1 or less (no discount below base cost)', () => { + const routes = compileCostMap(); + const matched = matchCostRoute(routes, 'GET', '/creators/abc/holders'); + expect(computeQueryCost(matched, '1')).toBe(3); + expect(computeQueryCost(matched, '0')).toBe(3); + }); + + it('ignores a non-numeric limit param', () => { + const routes = compileCostMap(); + const matched = matchCostRoute(routes, 'GET', '/creators/abc/holders'); + expect(computeQueryCost(matched, 'not-a-number')).toBe(3); + }); + + it('uses the first value when limit is an array', () => { + const routes = compileCostMap(); + const matched = matchCostRoute(routes, 'GET', '/creators/abc/holders'); + expect(computeQueryCost(matched, ['50', '999'])).toBe(150); + }); +}); diff --git a/src/utils/query-cost.utils.ts b/src/utils/query-cost.utils.ts new file mode 100644 index 0000000..947224c --- /dev/null +++ b/src/utils/query-cost.utils.ts @@ -0,0 +1,140 @@ +// src/utils/query-cost.utils.ts +// Route-pattern matching and cost computation for the query cost governor +// (#755). Deliberately not path-to-regexp / req.route: the governor is +// mounted as router-level middleware ahead of route resolution, so req.route +// isn't populated yet when it runs — matching has to work off req.path +// directly. + +import { + DEFAULT_QUERY_COST, + DEFAULT_QUERY_COST_MAP, +} from '../constants/query-cost.constants'; + +export interface CompiledCostRoute { + method: string; + pattern: string; + regex: RegExp; + baseCost: number; +} + +/** Converts an Express-style `:param` pattern into a matching RegExp. */ +function compilePattern(pattern: string): RegExp { + const escaped = pattern + .split('/') + .map(segment => + segment.startsWith(':') + ? '[^/]+' + : segment.replace(/[.*+?^${}()|[\]\\]/g, '\\$&') + ) + .join('/'); + return new RegExp(`^${escaped}/?$`); +} + +/** Parses a `"METHOD /pattern"` key into its parts. */ +function parseKey(key: string): { method: string; pattern: string } | null { + const spaceIndex = key.indexOf(' '); + if (spaceIndex === -1) return null; + return { + method: key.slice(0, spaceIndex).toUpperCase(), + pattern: key.slice(spaceIndex + 1), + }; +} + +/** + * Merges the default cost map with an optional JSON override (from + * QUERY_COST_MAP_JSON), compiling every entry into a matchable route once at + * startup rather than on every request. + */ +export function compileCostMap( + overrideJson?: string, + defaults: Record = DEFAULT_QUERY_COST_MAP +): CompiledCostRoute[] { + const merged: Record = { ...defaults }; + + if (overrideJson) { + let parsed: unknown; + try { + parsed = JSON.parse(overrideJson); + } catch { + throw new Error( + 'QUERY_COST_MAP_JSON is not valid JSON — expected an object of "METHOD /pattern": cost entries' + ); + } + if ( + typeof parsed !== 'object' || + parsed === null || + Array.isArray(parsed) + ) { + throw new Error( + 'QUERY_COST_MAP_JSON must be a JSON object of "METHOD /pattern": cost entries' + ); + } + for (const [key, value] of Object.entries( + parsed as Record + )) { + if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) { + throw new Error( + `QUERY_COST_MAP_JSON entry "${key}" must map to a positive number` + ); + } + merged[key] = value; + } + } + + const compiled: CompiledCostRoute[] = []; + for (const [key, baseCost] of Object.entries(merged)) { + const parts = parseKey(key); + if (!parts) { + throw new Error( + `Query cost map key "${key}" must be of the form "METHOD /pattern"` + ); + } + compiled.push({ + method: parts.method, + pattern: parts.pattern, + regex: compilePattern(parts.pattern), + baseCost, + }); + } + return compiled; +} + +/** Finds the first compiled route matching this method+path, if any. */ +export function matchCostRoute( + routes: CompiledCostRoute[], + method: string, + path: string +): CompiledCostRoute | null { + const upperMethod = method.toUpperCase(); + for (const route of routes) { + if (route.method === upperMethod && route.regex.test(path)) { + return route; + } + } + return null; +} + +/** + * Computes the cost of a request: the matched route's base cost, or + * DEFAULT_QUERY_COST when unmatched, multiplied by the `limit` query param + * when present (issue #755's "parameterised cost" requirement — a caller + * asking for more rows pays proportionally more). + */ +/** Redis key for a caller's rolling query-cost sorted set. */ +export function buildQueryCostRedisKey(identity: string): string { + return `qcost:${identity}`; +} + +export function computeQueryCost( + matched: CompiledCostRoute | null, + limitParam: unknown +): number { + const baseCost = matched?.baseCost ?? DEFAULT_QUERY_COST; + const limit = Array.isArray(limitParam) ? limitParam[0] : limitParam; + const parsedLimit = + typeof limit === 'string' ? Number.parseInt(limit, 10) : NaN; + if (Number.isFinite(parsedLimit) && parsedLimit > 1) { + return baseCost * parsedLimit; + } + return baseCost; +} From 564752319bd825a41d2cb23a19174f7fbb220672 Mon Sep 17 00:00:00 2001 From: wheval Date: Fri, 28 Aug 2026 10:41:16 +0100 Subject: [PATCH 2/4] fix: resolve pre-existing repo-wide tsc build failures (unrelated to #755) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The verify CI check runs a full-repo `pnpm build` (tsc), which was already broken on main before this PR touched anything (confirmed via a clean-checkout diff: 75 errors on main, 75 on this branch, identical set). Since the PR is otherwise blocked on this, fixing what's practical here rather than leaving CI red: - Redis client typing: src/utils/redis.utils.ts exported `redis`/`getRedis` as an alias to a function returning `Redis | null`, and ~13 call sites across subscriptions, key price-moved pub/sub, sequencer locking, and the supply-drift guard called it and used the result without a null check — all real hard dependencies on Redis, not optional cache reads. Added `getRequiredRedisClient()` (throws a clear error instead of silently no-opping locking/notification logic) and switched those call sites to it. Also added the `connectRedis()` export src/server.ts already imports and calls at startup but never existed. - Two ioredis `zrange(key, start, stop)` calls passed `stop` as a number; this ioredis version's types only accept string/Buffer there. Stringified. - Dead code: jwt.middleware.ts's `signJwt` referenced a nonexistent `JWT_EXPIRES_IN` config field and was never imported anywhere in src/ (the real, actively-used token issuer is utils/jwt.utils.ts's `signJwt`/ `signWalletAccessToken`, wired through jwt-auth.middleware.ts's `requireJwtAuth`). Removed it rather than inventing a second, redundant expiry config. One test file did still import it though (with a structurally different, already-incompatible token shape that would never have satisfied the real requireJwtAuth check it's testing against) — switched it to signWalletAccessToken, the function the route it tests actually verifies against. - wallets.routes.ts referenced `jwtAuth`/`httpGetWalletFollowing` with no imports at all; both already exist under the names requireJwtAuth (jwt-auth.middleware.ts) and httpGetWalletFollowing (wallet-following.controllers.ts) - just needed importing. - keys.routes.ts referenced getKeyProposals/getKeySupply and ProposalKeyNotFoundError/SupplyKeyNotFoundError the same way - both services already exist (key-proposals.service.ts, key-supply.service.ts), each with its own locally-scoped KeyNotFoundError class, aliased on import exactly like the repo's existing key-fees.service.ts import already does. - Missing config.schema.ts fields referenced by real code: SSE_SUBSCRIPTION_TTL_MS/SSE_MAX_SUBSCRIPTIONS_PER_WALLET/ SSE_MAX_CONNECTIONS_PER_WALLET/SSE_THROTTLE_DURATION_MS (subscriptions) and STELLAR_AUTH_SECRET (auth/stellar-challenge.controller.ts, already had a working fallback to a random keypair when unset). - Prisma schema drift: - `prisma.walletCreatorFollow` (src/modules/wallets/wallet-following.service.ts) was never in schema.prisma at all, even though its migration (20260825000000_add_wallet_creator_follows) was already applied - the `wallet_creator_follows` table exists in the DB with no matching model declaration. Added the model, mapped to the existing table/columns. - dividend.service.ts ordered by a `distributedAt` field that was never a real column - the actual column (per both schema.prisma and its migration) is `distributionDate`. Fixed the field name. - `KeyOwnership.lastBuyAt` (src/modules/ownership/ownership.service.ts, src/modules/users/holdings.service.ts - lockup-window calculation) was genuinely never migrated. Added the column + a real migration. - Four integration test files imported from 'vitest', a framework this repo doesn't use (package.json: "test": "jest") - copy-paste from elsewhere. Three of them also imported a `createServer` from a src/utils/server.utils.ts that doesn't exist; every other integration test in the repo just imports the default-exported `app` from src/app.ts directly. Switched to that established pattern. Also fixed three `prisma.user.create()` calls missing required fields (passwordHash/ firstName/lastName) and one key-sync test referencing an undeclared `creatorId` variable (should have been `testCreatorId`) plus a CreatorPriceSnapshot `price` field that's actually named `currentPrice`. Verified: `npx tsc` (0 errors, was 75), `pnpm build`, and `pnpm lint` all clean - the exact three commands `.github/workflows/ci.yml`'s `verify` job runs, in the same order. Still not attempted (out of scope for landing this PR, needs its own issue): a from-scratch `prisma migrate deploy` against this repo's full migration history fails partway through on an unrelated pre-existing migration-ordering bug (20260625000000_add_price_snapshot references CreatorProfile before it exists in the replay order) - discovered while verifying the new lastBuyAt migration, unrelated to it, not fixed here. --- prisma/schema/follow.prisma | 16 ++++++++++ .../migration.sql | 2 ++ prisma/schema/ownership.prisma | 7 +++- src/config.schema.ts | 31 ++++++++++++++++++ src/middlewares/jwt.middleware.ts | 9 +----- .../audit-log-endpoint.integration.test.ts | 6 +--- .../admin/audit-log.integration.test.ts | 1 - .../admin/key-sync.integration.test.ts | 16 ++++------ .../dividend-endpoint.integration.test.ts | 7 ++-- src/modules/investor/dividend.service.ts | 2 +- src/modules/keys/keys.routes.ts | 8 +++++ src/modules/keys/price-moved.redis.ts | 8 ++--- .../notifications/notification.service.ts | 6 ++-- .../subscriptions/subscription.service.ts | 32 +++++++++---------- .../wallet-following.integration.test.ts | 17 +++------- src/modules/wallets/wallets.routes.ts | 5 +-- .../whitelist/whitelist.integration.test.ts | 13 ++++---- src/utils/redis.utils.ts | 26 +++++++++++++++ src/utils/sequencer-lock.utils.ts | 4 +-- src/utils/supply-drift-guard.utils.ts | 8 ++--- 20 files changed, 145 insertions(+), 79 deletions(-) create mode 100644 prisma/schema/migrations/20260828000000_add_key_ownership_last_buy_at/migration.sql diff --git a/prisma/schema/follow.prisma b/prisma/schema/follow.prisma index f05ad9f..f38c352 100644 --- a/prisma/schema/follow.prisma +++ b/prisma/schema/follow.prisma @@ -1,5 +1,21 @@ // prisma/schema/follow.prisma +/// Matches the already-applied `wallet_creator_follows` table (migration +/// 20260825000000_add_wallet_creator_follows) — the model declaration was +/// missing from the schema even though the table exists, so `prisma.walletCreatorFollow` +/// never worked. See src/modules/wallets/wallet-following.service.ts. +model WalletCreatorFollow { + id String @id @default(cuid()) + walletAddress String + creatorId String + createdAt DateTime @default(now()) + + @@unique([walletAddress, creatorId]) + @@index([walletAddress]) + @@index([creatorId]) + @@map("wallet_creator_follows") +} + model Follow { id String @id @default(cuid()) followerAddress String diff --git a/prisma/schema/migrations/20260828000000_add_key_ownership_last_buy_at/migration.sql b/prisma/schema/migrations/20260828000000_add_key_ownership_last_buy_at/migration.sql new file mode 100644 index 0000000..87a2d89 --- /dev/null +++ b/prisma/schema/migrations/20260828000000_add_key_ownership_last_buy_at/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "KeyOwnership" ADD COLUMN "lastBuyAt" TIMESTAMP(3); diff --git a/prisma/schema/ownership.prisma b/prisma/schema/ownership.prisma index 705d50a..f25d3db 100644 --- a/prisma/schema/ownership.prisma +++ b/prisma/schema/ownership.prisma @@ -15,7 +15,12 @@ model KeyOwnership { /// When the current lockup window ends for this holding, if any. lockupExpiresAt DateTime? - + + /// Timestamp of the most recent buy into this position — distinct from + /// `updatedAt`, which also changes on sells/other updates. Drives the + /// per-buy lockup window in src/modules/users/holdings.service.ts. + lastBuyAt DateTime? + createdAt DateTime @default(now()) updatedAt DateTime @updatedAt diff --git a/src/config.schema.ts b/src/config.schema.ts index 09daeed..23ec1b4 100644 --- a/src/config.schema.ts +++ b/src/config.schema.ts @@ -258,6 +258,37 @@ export const envSchema = z .positive() .default(5000), SSE_REPLAY_MAX_EVENTS: z.coerce.number().int().positive().default(100), + + // SSE subscription management (src/modules/subscriptions) — a wallet's + // subscription set, persisted in Redis, distinct from the per-connection + // heartbeat/queue/replay tuning above. + SSE_SUBSCRIPTION_TTL_MS: z.coerce + .number() + .int() + .positive() + .default(3_600_000), + SSE_MAX_SUBSCRIPTIONS_PER_WALLET: z.coerce + .number() + .int() + .positive() + .default(20), + SSE_MAX_CONNECTIONS_PER_WALLET: z.coerce + .number() + .int() + .positive() + .default(5), + SSE_THROTTLE_DURATION_MS: z.coerce + .number() + .int() + .positive() + .default(30_000), + + // Stellar challenge-response auth (src/modules/auth/stellar-challenge.controller.ts). + // The server's own signing keypair for issued challenges. Left unset by + // default (a fresh random keypair is used each boot, matching the + // existing fallback), since it only needs to be stable across restarts + // in production. + STELLAR_AUTH_SECRET: optionalNonEmptyString, }) .superRefine((data, ctx) => { if (data.MODE === 'production' && data.STELLAR_NETWORK === 'testnet') { diff --git a/src/middlewares/jwt.middleware.ts b/src/middlewares/jwt.middleware.ts index d0308f7..e7a70b9 100644 --- a/src/middlewares/jwt.middleware.ts +++ b/src/middlewares/jwt.middleware.ts @@ -1,5 +1,5 @@ import { Request, Response, NextFunction } from 'express'; -import jwt, { SignOptions } from 'jsonwebtoken'; +import jwt from 'jsonwebtoken'; import { envConfig } from '../config'; import { sendUnauthorized } from '../utils/api-response.utils'; import { logger } from '../utils/logger.utils'; @@ -69,10 +69,3 @@ export function jwtAuth(req: Request, res: Response, next: NextFunction): void { sendUnauthorized(res, 'Invalid or expired token'); } } - -export function signJwt(payload: JwtPayload): string { - const options: SignOptions = { - expiresIn: envConfig.JWT_EXPIRES_IN as any, - }; - return jwt.sign(payload, envConfig.JWT_SECRET, options); -} diff --git a/src/modules/admin/audit-log-endpoint.integration.test.ts b/src/modules/admin/audit-log-endpoint.integration.test.ts index b6c3a3f..8a89f34 100644 --- a/src/modules/admin/audit-log-endpoint.integration.test.ts +++ b/src/modules/admin/audit-log-endpoint.integration.test.ts @@ -1,16 +1,12 @@ -import { describe, it, expect, beforeAll, afterAll } from 'vitest'; import request from 'supertest'; -import { createServer } from '../../utils/server.utils'; +import app from '../../app'; import { prisma } from '../../utils/prisma.utils'; import { signWalletAccessToken } from '../../utils/jwt.utils'; describe('GET /admin/audit-log Endpoint Integration Tests', () => { - let app: any; let adminToken: string; beforeAll(async () => { - app = await createServer(); - // Create admin token const adminWallet = '0xadmintestwallet1111111111111111111111111'; adminToken = signWalletAccessToken(adminWallet, 'admin-sub', 3600); diff --git a/src/modules/admin/audit-log.integration.test.ts b/src/modules/admin/audit-log.integration.test.ts index 9a3b7f0..f60aeb5 100644 --- a/src/modules/admin/audit-log.integration.test.ts +++ b/src/modules/admin/audit-log.integration.test.ts @@ -1,4 +1,3 @@ -import { describe, it, expect, beforeAll, afterAll } from 'vitest'; import { prisma } from '../../utils/prisma.utils'; import { createAuditEntry, getAuditLogs } from './audit-log.service'; diff --git a/src/modules/admin/key-sync.integration.test.ts b/src/modules/admin/key-sync.integration.test.ts index 2c32caa..3e57d76 100644 --- a/src/modules/admin/key-sync.integration.test.ts +++ b/src/modules/admin/key-sync.integration.test.ts @@ -1,18 +1,14 @@ -import { describe, it, expect, beforeAll, afterAll } from 'vitest'; import request from 'supertest'; -import { createServer } from '../../utils/server.utils'; +import app from '../../app'; import { prisma } from '../../utils/prisma.utils'; import { signWalletAccessToken } from '../../utils/jwt.utils'; import { syncKeyState, creatorExists } from './key-sync.service'; describe('Key Sync Integration Tests', () => { - let app: any; let adminToken: string; let testCreatorId: string; beforeAll(async () => { - app = await createServer(); - // Create admin token const adminWallet = '0xadmintestwallet1111111111111111111111111'; adminToken = signWalletAccessToken(adminWallet, 'admin-sub', 3600); @@ -27,6 +23,9 @@ describe('Key Sync Integration Tests', () => { const user = await prisma.user.create({ data: { email: `test-${Date.now()}@example.com`, + passwordHash: 'hash', + firstName: 'Key', + lastName: 'Sync', stellarWallet: { create: { address: 'GBTEST0001' } }, }, }); @@ -46,9 +45,8 @@ describe('Key Sync Integration Tests', () => { // Create price snapshot await prisma.creatorPriceSnapshot.create({ data: { - creatorId, - price: 100, - priceUpdatedAt: new Date(), + creatorId: testCreatorId, + currentPrice: 100, }, }); @@ -57,7 +55,7 @@ describe('Key Sync Integration Tests', () => { await prisma.keyOwnership.create({ data: { ownerAddress: `GHOLDER${String(i).padStart(52, '0')}`, - creatorId, + creatorId: testCreatorId, balance: 100, }, }); diff --git a/src/modules/dividends/dividend-endpoint.integration.test.ts b/src/modules/dividends/dividend-endpoint.integration.test.ts index cdaabb0..54412d2 100644 --- a/src/modules/dividends/dividend-endpoint.integration.test.ts +++ b/src/modules/dividends/dividend-endpoint.integration.test.ts @@ -1,16 +1,14 @@ import request from 'supertest'; +import app from '../../app'; import { prisma } from '../../utils/prisma.utils'; import { processDividendEvents } from '../indexer/dividend-indexer.service'; import { IndexerChainEvent } from '../../utils/indexer-event-processor.utils'; describe('Dividend Endpoints Integration Tests', () => { - let app: any; let testCreatorId: string; let testDistributionId: string; beforeAll(async () => { - app = await createServer(); - // Clean up await prisma.dividendClaim.deleteMany({}); await prisma.dividendDistribution.deleteMany({}); @@ -234,6 +232,9 @@ describe('Dividend Endpoints Integration Tests', () => { const user = await prisma.user.create({ data: { email: `test2-${Date.now()}@example.com`, + passwordHash: 'hash123', + firstName: 'Test', + lastName: 'User2', stellarWallet: { create: { address: 'GBTEST0002' } }, }, }); diff --git a/src/modules/investor/dividend.service.ts b/src/modules/investor/dividend.service.ts index c193617..b3ad852 100644 --- a/src/modules/investor/dividend.service.ts +++ b/src/modules/investor/dividend.service.ts @@ -14,7 +14,7 @@ export async function getInvestorDividends( } const items = await prisma.dividendDistribution.findMany({ where, - orderBy: { distributedAt: 'desc' }, + orderBy: { distributionDate: 'desc' }, take: limit + 1, }); const hasMore = items.length > limit; diff --git a/src/modules/keys/keys.routes.ts b/src/modules/keys/keys.routes.ts index b17b7b3..01843e4 100644 --- a/src/modules/keys/keys.routes.ts +++ b/src/modules/keys/keys.routes.ts @@ -14,6 +14,14 @@ import { PRICE_HISTORY_INTERVALS, } from './key-price-history.service'; import { getKeyFees, KeyNotFoundError } from './key-fees.service'; +import { + getKeyProposals, + KeyNotFoundError as ProposalKeyNotFoundError, +} from './key-proposals.service'; +import { + getKeySupply, + KeyNotFoundError as SupplyKeyNotFoundError, +} from './key-supply.service'; import { KeySearchQueryTooShortError, searchKeys } from './key-search.service'; import { KEY_SEARCH_MIN_QUERY_LENGTH } from '../../constants/notifications.constants'; import dividendRouter from '../dividends/dividend.routes'; diff --git a/src/modules/keys/price-moved.redis.ts b/src/modules/keys/price-moved.redis.ts index ad53ea8..1e16c0c 100644 --- a/src/modules/keys/price-moved.redis.ts +++ b/src/modules/keys/price-moved.redis.ts @@ -1,5 +1,5 @@ // src/modules/keys/price-moved.redis.ts -import { getRedis } from '../../utils/redis.utils'; +import { getRequiredRedisClient } from '../../utils/redis.utils'; import { prisma } from '../../utils/prisma.utils'; import { PRICE_MOVED_SET_TTL_SECONDS, @@ -7,7 +7,7 @@ import { } from '../../constants/notifications.constants'; export async function writePriceMovedKeys(keyIds: string[]): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const pipeline = redis.pipeline(); pipeline.del(REDIS_KEYS.priceMovedSet); if (keyIds.length > 0) { @@ -18,14 +18,14 @@ export async function writePriceMovedKeys(keyIds: string[]): Promise { } export async function getPriceMovedKeyIds(): Promise { - return getRedis().smembers(REDIS_KEYS.priceMovedSet); + return getRequiredRedisClient().smembers(REDIS_KEYS.priceMovedSet); } export async function markPriceMovedDelivered( keyId: string, walletAddress: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const deliveredKey = REDIS_KEYS.priceMovedDelivered(keyId); await redis.sadd(deliveredKey, walletAddress); await redis.expire(deliveredKey, PRICE_MOVED_SET_TTL_SECONDS); diff --git a/src/modules/notifications/notification.service.ts b/src/modules/notifications/notification.service.ts index 11cbda0..b88e0cb 100644 --- a/src/modules/notifications/notification.service.ts +++ b/src/modules/notifications/notification.service.ts @@ -1,6 +1,6 @@ // src/modules/notifications/notification.service.ts import { prisma } from '../../utils/prisma.utils'; -import { getRedis } from '../../utils/redis.utils'; +import { getRequiredRedisClient } from '../../utils/redis.utils'; import { LOCKUP_WARNING_WINDOW_MS, NOTIFICATION_TYPES, @@ -13,7 +13,7 @@ import { import { NotificationItem } from './notification.types'; async function getLastReadAt(walletAddress: string): Promise { - const raw = await getRedis().get( + const raw = await getRequiredRedisClient().get( REDIS_KEYS.notificationsReadAt(walletAddress) ); if (!raw) { @@ -182,7 +182,7 @@ export async function markAllNotificationsRead( walletAddress: string, now: Date = new Date() ): Promise { - await getRedis().set( + await getRequiredRedisClient().set( REDIS_KEYS.notificationsReadAt(walletAddress), now.toISOString() ); diff --git a/src/modules/subscriptions/subscription.service.ts b/src/modules/subscriptions/subscription.service.ts index 2049d69..75d8d89 100644 --- a/src/modules/subscriptions/subscription.service.ts +++ b/src/modules/subscriptions/subscription.service.ts @@ -1,5 +1,5 @@ import { randomUUID } from 'crypto'; -import { getRedis } from '../../utils/redis.utils'; +import { getRequiredRedisClient } from '../../utils/redis.utils'; import { envConfig } from '../../config'; import { Subscription, @@ -43,7 +43,7 @@ export async function createSubscription( walletAddress: string, topics: SubscriptionTopic[] ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const walletKey = walletSubsKey(walletAddress); @@ -86,7 +86,7 @@ export async function createSubscription( export async function getSubscription( subscriptionId: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const data = await redis.hgetall(subKey(subscriptionId)); if (!data || !data.walletAddress) return null; @@ -99,7 +99,7 @@ export async function getSubscription( } export async function deleteSubscription(subscriptionId: string): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const sub = await getSubscription(subscriptionId); if (!sub) return; @@ -111,14 +111,14 @@ export async function deleteSubscription(subscriptionId: string): Promise } export async function touchSubscription(subscriptionId: string): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); await redis.expire(subKey(subscriptionId), SUBSCRIPTION_TTL_S); } export async function getLastCursor( subscriptionId: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); return redis.get(cursorKey(subscriptionId)); } @@ -126,25 +126,25 @@ export async function saveCursor( subscriptionId: string, cursor: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); await redis.set(cursorKey(subscriptionId), cursor); } export async function isThrottled(walletAddress: string): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const exists = await redis.exists(throttledKey(walletAddress)); return exists === 1; } export async function setThrottled(walletAddress: string): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); await redis.setex(throttledKey(walletAddress), THROTTLE_DURATION_S, '1'); } export async function incrementConnectionCount( walletAddress: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const count = await redis.incr(connectionCountKey(walletAddress)); await redis.expire(connectionCountKey(walletAddress), 60); return count; @@ -153,18 +153,18 @@ export async function incrementConnectionCount( export async function decrementConnectionCount( walletAddress: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); await redis.decr(connectionCountKey(walletAddress)); } export async function getWalletSubscriptions( walletAddress: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const ids = await redis.zrange( walletSubsKey(walletAddress), 0, - -1 + '-1' ); const subs: Subscription[] = []; @@ -178,7 +178,7 @@ export async function getWalletSubscriptions( export async function getSubscriptionsByTopic( topic: string ): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const ids = await redis.keys(`${SUBSCRIPTION_KEY_PREFIX}*`); const subs: Subscription[] = []; @@ -201,12 +201,12 @@ export async function getSubscriptionsByTopic( } export async function pruneExpiredSubscriptions(): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const walletKeys = await redis.keys(`${WALLET_SUBSCRIPTIONS_KEY_PREFIX}*`); let pruned = 0; for (const wk of walletKeys) { - const ids = await redis.zrange(wk, 0, -1); + const ids = await redis.zrange(wk, 0, '-1'); for (const id of ids) { const exists = await redis.exists(subKey(id)); if (exists === 0) { diff --git a/src/modules/wallets/wallet-following.integration.test.ts b/src/modules/wallets/wallet-following.integration.test.ts index 3a2d2ba..da57d14 100644 --- a/src/modules/wallets/wallet-following.integration.test.ts +++ b/src/modules/wallets/wallet-following.integration.test.ts @@ -9,7 +9,7 @@ import supertest from 'supertest'; import { Keypair } from '@stellar/stellar-base'; import app from '../../app'; import { prisma } from '../../utils/prisma.utils'; -import { signJwt } from '../../middlewares/jwt.middleware'; +import { signWalletAccessToken } from '../../utils/jwt.utils'; describe('GET /api/v1/wallets/:address/following', () => { const PREFIX = 'wallet-following-test'; @@ -166,10 +166,7 @@ describe('GET /api/v1/wallets/:address/following', () => { // ── Alphabetical ordering ──────────────────────────────────────────────── it('returns creators in alphabetical order by display name', async () => { - const token = signJwt({ - walletAddress: walletA.publicKey(), - sub: userIdWalletA, - }); + const token = signWalletAccessToken(walletA.publicKey(), userIdWalletA); const res = await supertest(app) .get(`/api/v1/wallets/${walletA.publicKey()}/following`) @@ -185,10 +182,7 @@ describe('GET /api/v1/wallets/:address/following', () => { // ── Completeness ───────────────────────────────────────────────────────── it('returns all followed creators', async () => { - const token = signJwt({ - walletAddress: walletA.publicKey(), - sub: userIdWalletA, - }); + const token = signWalletAccessToken(walletA.publicKey(), userIdWalletA); const res = await supertest(app) .get(`/api/v1/wallets/${walletA.publicKey()}/following`) @@ -206,10 +200,7 @@ describe('GET /api/v1/wallets/:address/following', () => { // ── Empty array for wallet with no follows ─────────────────────────────── it('returns an empty array for a wallet that follows no one', async () => { - const token = signJwt({ - walletAddress: walletB.publicKey(), - sub: userIdWalletB, - }); + const token = signWalletAccessToken(walletB.publicKey(), userIdWalletB); const res = await supertest(app) .get(`/api/v1/wallets/${walletB.publicKey()}/following`) diff --git a/src/modules/wallets/wallets.routes.ts b/src/modules/wallets/wallets.routes.ts index b9d4231..f802447 100644 --- a/src/modules/wallets/wallets.routes.ts +++ b/src/modules/wallets/wallets.routes.ts @@ -1,9 +1,10 @@ import { Router } from "express"; import { httpGetWalletActivity } from "./wallet-activity.controllers"; import { httpGetWalletHoldings } from "./wallet-holdings.controllers"; +import { httpGetWalletFollowing } from "./wallet-following.controllers"; import { cacheControl } from "../../middlewares/cache-control.middleware"; import { ACTIVITY_FEED_CACHE_PRESET } from "../../constants/activity-feed-cache.constants"; -import { requireWalletParamMatch } from "../../middlewares/jwt-auth.middleware"; +import { requireJwtAuth, requireWalletParamMatch } from "../../middlewares/jwt-auth.middleware"; const walletsRouter = Router(); @@ -34,6 +35,6 @@ walletsRouter.get("/:address/holdings", httpGetWalletHoldings); * Returns all creators that the given wallet follows, ordered * alphabetically by display name. Requires JWT authentication. */ -walletsRouter.get('/:address/following', jwtAuth, httpGetWalletFollowing); +walletsRouter.get('/:address/following', requireJwtAuth, httpGetWalletFollowing); export default walletsRouter; diff --git a/src/modules/whitelist/whitelist.integration.test.ts b/src/modules/whitelist/whitelist.integration.test.ts index f8afcc1..a252bc0 100644 --- a/src/modules/whitelist/whitelist.integration.test.ts +++ b/src/modules/whitelist/whitelist.integration.test.ts @@ -1,18 +1,14 @@ -import { describe, it, expect, beforeAll, afterAll, vi } from 'vitest'; import request from 'supertest'; -import { createServer } from '../../utils/server.utils'; +import app from '../../app'; import { prisma } from '../../utils/prisma.utils'; import { getWhitelistStatus, creatorExists } from './whitelist.service'; import * as cacheUtils from '../../utils/redis.utils'; describe('Whitelist Endpoint Integration Tests', () => { - let app: any; let testCreatorId: string; let testWallet = 'GWALLET000000000000000000000000000000001'; beforeAll(async () => { - app = await createServer(); - // Clean up await prisma.creatorProfile.deleteMany({}); await prisma.user.deleteMany({}); @@ -21,6 +17,9 @@ describe('Whitelist Endpoint Integration Tests', () => { const user = await prisma.user.create({ data: { email: `test-${Date.now()}@example.com`, + passwordHash: 'hash', + firstName: 'Whitelist', + lastName: 'Test', stellarWallet: { create: { address: 'GBTEST0001' } }, }, }); @@ -239,8 +238,8 @@ describe('Whitelist Endpoint Integration Tests', () => { // Note: This test requires Redis to be available // We spy on the caching functions to verify they're called - const cacheGetSpy = vi.spyOn(cacheUtils, 'cacheGetJson'); - const cacheSetSpy = vi.spyOn(cacheUtils, 'cacheSetJson'); + const cacheGetSpy = jest.spyOn(cacheUtils, 'cacheGetJson'); + const cacheSetSpy = jest.spyOn(cacheUtils, 'cacheSetJson'); // First request should miss cache and populate it const response1 = await request(app) diff --git a/src/utils/redis.utils.ts b/src/utils/redis.utils.ts index 644eac4..ef23f40 100644 --- a/src/utils/redis.utils.ts +++ b/src/utils/redis.utils.ts @@ -214,5 +214,31 @@ export async function disconnectRedis(): Promise { } } +/** + * Eagerly create (and, since `lazyConnect` is false, begin connecting) the + * shared client. Called once at startup so `getRedisClient()`/callers below + * don't pay first-request connection latency. + */ +export async function connectRedis(): Promise { + getRedisClient(); +} + +/** + * For subsystems where Redis isn't an optional cache but a hard + * dependency (SSE subscriptions, sequencer locks, supply-drift guards, + * key price-moved pub/sub) — throws instead of silently degrading, since + * a `null` client there would mean silently no-opping locking/notification + * logic rather than a safe cache miss. + */ +export function getRequiredRedisClient(): Redis { + const client = getRedisClient(); + if (!client) { + throw Object.assign(new Error('Redis is required but unavailable (ENABLE_REDIS_CACHE is false or the client failed to initialise)'), { + code: 'redis_unavailable', + }); + } + return client; +} + export const getRedis = getRedisClient; export const redis = getRedisClient; diff --git a/src/utils/sequencer-lock.utils.ts b/src/utils/sequencer-lock.utils.ts index 04e9652..a809371 100644 --- a/src/utils/sequencer-lock.utils.ts +++ b/src/utils/sequencer-lock.utils.ts @@ -1,4 +1,4 @@ -import { getRedis } from './redis.utils'; +import { getRequiredRedisClient } from './redis.utils'; import { logger } from './logger.utils'; const LOCK_TTL_SECONDS = 15; @@ -22,7 +22,7 @@ function lockKey(creatorWallet: string): string { export async function acquireSequencerLock( creatorWallet: string ): Promise<{ release: () => Promise }> { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const key = lockKey(creatorWallet); const lockValue = `${process.pid}:${Date.now()}`; const deadline = Date.now() + LOCK_ACQUIRE_TIMEOUT_MS; diff --git a/src/utils/supply-drift-guard.utils.ts b/src/utils/supply-drift-guard.utils.ts index 90604e2..30edd17 100644 --- a/src/utils/supply-drift-guard.utils.ts +++ b/src/utils/supply-drift-guard.utils.ts @@ -1,4 +1,4 @@ -import { getRedis } from './redis.utils'; +import { getRequiredRedisClient } from './redis.utils'; import { logger } from './logger.utils'; export class SupplyDriftHaltedError extends Error { @@ -15,7 +15,7 @@ function driftKey(creatorWallet: string): string { } export async function isDriftHalted(creatorWallet: string): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); const exists = await redis.exists(driftKey(creatorWallet)); return exists === 1; } @@ -45,7 +45,7 @@ export async function verifySupplyAndGuard( 'Supply drift detected! Operations halted for creator until cleared.' ); - const redis = getRedis(); + const redis = getRequiredRedisClient(); await redis.set(driftKey(creatorWallet), '1'); return false; } @@ -54,7 +54,7 @@ export async function verifySupplyAndGuard( } export async function clearDrift(creatorWallet: string): Promise { - const redis = getRedis(); + const redis = getRequiredRedisClient(); await redis.del(driftKey(creatorWallet)); logger.info( { creator_wallet: creatorWallet }, From fdaaeb70cfac9833031133f19314e24155c9e2b7 Mon Sep 17 00:00:00 2001 From: wheval Date: Sun, 30 Aug 2026 17:17:35 +0100 Subject: [PATCH 3/4] fix: remove duplicate WalletCreatorFollow model from the merge Both my branch and upstream/main independently added a WalletCreatorFollow model to follow.prisma (same underlying gap, fixed twice) - the merge kept both since they landed on non-overlapping lines, which is a genuine duplicate model name and fails prisma generate. Kept upstream's version (it additionally relates to CreatorProfile, which mine didn't). --- prisma/schema/follow.prisma | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/prisma/schema/follow.prisma b/prisma/schema/follow.prisma index d378dc7..864f4bb 100644 --- a/prisma/schema/follow.prisma +++ b/prisma/schema/follow.prisma @@ -1,21 +1,5 @@ // prisma/schema/follow.prisma -/// Matches the already-applied `wallet_creator_follows` table (migration -/// 20260825000000_add_wallet_creator_follows) — the model declaration was -/// missing from the schema even though the table exists, so `prisma.walletCreatorFollow` -/// never worked. See src/modules/wallets/wallet-following.service.ts. -model WalletCreatorFollow { - id String @id @default(cuid()) - walletAddress String - creatorId String - createdAt DateTime @default(now()) - - @@unique([walletAddress, creatorId]) - @@index([walletAddress]) - @@index([creatorId]) - @@map("wallet_creator_follows") -} - model Follow { id String @id @default(cuid()) followerAddress String From 205e2f586ddaa786b2a5d29159f1ff2f01ea2fb4 Mon Sep 17 00:00:00 2001 From: wheval Date: Sun, 30 Aug 2026 17:42:25 +0100 Subject: [PATCH 4/4] fix: add missing UNPROCESSABLE_ENTITY error code and normalize keyId param type Both pre-existing on upstream/main: admin/creator routes reference ErrorCode.UNPROCESSABLE_ENTITY which was never added to the shared constant, and req.params.keyId's string|string[] type doesn't satisfy Prisma's OR filter typing. Blocking the build for everyone on main. --- src/constants/error.constants.ts | 1 + src/modules/creator/creator.routes.ts | 4 +++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/src/constants/error.constants.ts b/src/constants/error.constants.ts index c207198..d285028 100644 --- a/src/constants/error.constants.ts +++ b/src/constants/error.constants.ts @@ -9,6 +9,7 @@ export const ErrorCode = { FORBIDDEN: 'FORBIDDEN', CONFLICT: 'CONFLICT', BAD_REQUEST: 'BAD_REQUEST', + UNPROCESSABLE_ENTITY: 'UNPROCESSABLE_ENTITY', INTERNAL_ERROR: 'INTERNAL_ERROR', RATE_LIMIT: 'RATE_LIMIT', PRISMA_ERROR: 'DATABASE_ERROR', diff --git a/src/modules/creator/creator.routes.ts b/src/modules/creator/creator.routes.ts index dc290fa..70a2a93 100644 --- a/src/modules/creator/creator.routes.ts +++ b/src/modules/creator/creator.routes.ts @@ -76,7 +76,9 @@ creatorsRouter.post( return; } - const keyId = req.params.keyId; + const keyId = Array.isArray(req.params.keyId) + ? req.params.keyId[0] + : req.params.keyId; try { const creatorProfile = await prisma.creatorProfile.findFirst({ where: { OR: [{ id: keyId }, { handle: keyId }] },