diff --git a/src/config.schema.ts b/src/config.schema.ts index 2b4781b..108f351 100644 --- a/src/config.schema.ts +++ b/src/config.schema.ts @@ -168,6 +168,23 @@ export const envSchema = z .default('https://soroban-testnet.stellar.org'), STELLAR_AUTH_SECRET: optionalNonEmptyString, + // Shared secret that lets trusted internal services bypass per-wallet + // rate limits (e.g. the buy endpoint's sliding window limiter). + // Unset by default — no requests bypass rate limiting until configured. + INTERNAL_SERVICE_KEY: optionalNonEmptyString, + + // Volume leaderboard (GET /api/v1/creators/leaderboard/volume, #785). + LEADERBOARD_VOLUME_WINDOW_DAYS: z.coerce + .number() + .int() + .positive() + .default(7), + LEADERBOARD_VOLUME_CACHE_TTL_SECONDS: z.coerce + .number() + .int() + .positive() + .default(300), + // Ownership snapshot cleanup job OWNERSHIP_SNAPSHOT_TABLE_NAME: z .string() diff --git a/src/middlewares/validate-body.middleware.test.ts b/src/middlewares/validate-body.middleware.test.ts new file mode 100644 index 0000000..5586daa --- /dev/null +++ b/src/middlewares/validate-body.middleware.test.ts @@ -0,0 +1,86 @@ +import { z } from 'zod'; +import { validateBody } from './validate-body.middleware'; + +function makeRes(): any { + const res: any = {}; + res.status = jest.fn().mockReturnValue(res); + res.json = jest.fn().mockReturnValue(res); + res.setHeader = jest.fn().mockReturnValue(res); + return res; +} + +const schema = z.object({ + name: z.string().min(1), + age: z.number().int().positive(), +}); + +describe('validateBody', () => { + it('calls next() and replaces req.body with the parsed data on success', () => { + const req: any = { body: { name: 'Ada', age: 30 } }; + const res = makeRes(); + const next = jest.fn(); + + validateBody(schema)(req, res, next); + + expect(next).toHaveBeenCalled(); + expect(res.status).not.toHaveBeenCalled(); + expect(req.body).toEqual({ name: 'Ada', age: 30 }); + }); + + it('strips unknown fields before reaching the controller', () => { + const req: any = { + body: { name: 'Ada', age: 30, isAdmin: true, extra: 'nope' }, + }; + const res = makeRes(); + const next = jest.fn(); + + validateBody(schema)(req, res, next); + + expect(next).toHaveBeenCalled(); + expect(req.body).toEqual({ name: 'Ada', age: 30 }); + expect(req.body).not.toHaveProperty('isAdmin'); + expect(req.body).not.toHaveProperty('extra'); + }); + + it('returns 422 with per-field details when a required field is missing', () => { + const req: any = { body: { age: 30 } }; + const res = makeRes(); + const next = jest.fn(); + + validateBody(schema)(req, res, next); + + expect(next).not.toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(422); + const body = res.json.mock.calls[0][0]; + expect(body.success).toBe(false); + expect(body.error.code).toBe('VALIDATION_ERROR'); + expect(body.error.details).toEqual( + expect.arrayContaining([ + expect.objectContaining({ field: 'name' }), + ]) + ); + }); + + it('returns 422 with the field name and message when a type is wrong', () => { + const req: any = { body: { name: 'Ada', age: 'not-a-number' } }; + const res = makeRes(); + const next = jest.fn(); + + validateBody(schema)(req, res, next); + + expect(res.status).toHaveBeenCalledWith(422); + const body = res.json.mock.calls[0][0]; + expect(body.error.details[0].field).toBe('age'); + expect(body.error.details[0].message).toEqual(expect.any(String)); + }); + + it('passes valid bodies through unchanged (no extraneous mutation)', () => { + const req: any = { body: { name: 'Grace', age: 42 } }; + const res = makeRes(); + const next = jest.fn(); + + validateBody(schema)(req, res, next); + + expect(req.body).toEqual({ name: 'Grace', age: 42 }); + }); +}); diff --git a/src/middlewares/validate-body.middleware.ts b/src/middlewares/validate-body.middleware.ts new file mode 100644 index 0000000..659daab --- /dev/null +++ b/src/middlewares/validate-body.middleware.ts @@ -0,0 +1,44 @@ +// src/middlewares/validate-body.middleware.ts +// Centralized Zod request body validation. +// +// Mount `validateBody(schema)` ahead of a route handler to validate +// `req.body` before it reaches business logic. Unknown fields are stripped +// (the default behavior of `z.object()`), and invalid payloads short-circuit +// with a structured 422 response instead of reaching the controller. + +import type { Request, Response, NextFunction } from 'express'; +import type { ZodTypeAny } from 'zod'; +import { + sendError, + zodIssuesToDetails, + ErrorCode, +} from '../utils/api-response.utils'; + +/** + * Builds middleware that validates `req.body` against `schema` via + * `safeParse`, replacing `req.body` with the parsed (and unknown-field + * stripped) result on success. + * + * On failure, responds 422 with `{ error: { code: VALIDATION_ERROR, details } }` + * where `details` lists every invalid field and its message — the handler is + * never invoked. + */ +export function validateBody(schema: ZodTypeAny) { + return (req: Request, res: Response, next: NextFunction): void => { + const result = schema.safeParse(req.body); + + if (!result.success) { + sendError( + res, + 422, + ErrorCode.VALIDATION_ERROR, + 'Invalid request body', + zodIssuesToDetails(result.error.issues) + ); + return; + } + + req.body = result.data; + next(); + }; +} diff --git a/src/middlewares/wallet-rate-limit.middleware.test.ts b/src/middlewares/wallet-rate-limit.middleware.test.ts new file mode 100644 index 0000000..8e72a72 --- /dev/null +++ b/src/middlewares/wallet-rate-limit.middleware.test.ts @@ -0,0 +1,209 @@ +// Unit tests for the per-wallet sliding-window rate limiter (#779). + +const mockEnvConfig: { INTERNAL_SERVICE_KEY?: string } = { + INTERNAL_SERVICE_KEY: undefined, +}; + +jest.mock('../config', () => ({ + envConfig: mockEnvConfig, +})); + +jest.mock('../utils/logger.utils', () => ({ + logger: { + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + info: jest.fn(), + }, +})); + +type PipelineCommand = [string, ...unknown[]]; + +function buildFakeRedis(initialCount = 0) { + const commands: PipelineCommand[] = []; + let currentCount = initialCount; + + const pipeline = { + zremrangebyscore: (...args: unknown[]) => { + commands.push(['zremrangebyscore', ...args]); + return pipeline; + }, + zadd: (...args: unknown[]) => { + commands.push(['zadd', ...args]); + currentCount += 1; + return pipeline; + }, + zcard: (...args: unknown[]) => { + commands.push(['zcard', ...args]); + return pipeline; + }, + pexpire: (...args: unknown[]) => { + commands.push(['pexpire', ...args]); + return pipeline; + }, + exec: jest.fn(async () => [ + [null, 0], // zremrangebyscore + [null, 1], // zadd + [null, currentCount], // zcard + [null, 1], // pexpire + ]), + }; + + return { + pipeline: jest.fn(() => pipeline), + __setCount: (n: number) => { + currentCount = n; + }, + }; +} + +jest.mock('../utils/redis.utils', () => ({ + getRedis: jest.fn(), +})); + +import { getRedis } from '../utils/redis.utils'; +import { walletRateLimit } from './wallet-rate-limit.middleware'; +import type { StellarSignedRequest } from './stellar-signature.middleware'; + +const mockGetRedis = getRedis as jest.Mock; + +function makeReq(walletAddress?: string, headers: Record = {}) { + return { + walletAddress, + headers, + path: '/api/v1/creators/creator-1/buy', + } as unknown as StellarSignedRequest; +} + +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('walletRateLimit', () => { + beforeEach(() => { + mockEnvConfig.INTERNAL_SERVICE_KEY = undefined; + jest.clearAllMocks(); + }); + + it('allows the request when the wallet is under the limit', async () => { + const fakeRedis = buildFakeRedis(3); + mockGetRedis.mockReturnValue(fakeRedis); + const middleware = walletRateLimit({ + windowMs: 10_000, + max: 5, + keyPrefix: 'rl:test:', + }); + + const req = makeReq('GBUYER'); + const res = makeRes(); + const next = jest.fn(); + await middleware(req, res, next); + + expect(next).toHaveBeenCalled(); + expect(res.status).not.toHaveBeenCalled(); + }); + + it('returns 429 with Retry-After when the wallet exceeds the limit', async () => { + const fakeRedis = buildFakeRedis(6); + mockGetRedis.mockReturnValue(fakeRedis); + const middleware = walletRateLimit({ + windowMs: 10_000, + max: 5, + keyPrefix: 'rl:test:', + }); + + const req = makeReq('GBUYER'); + const res = makeRes(); + const next = jest.fn(); + await middleware(req, res, next); + + expect(next).not.toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(429); + expect(res.set).toHaveBeenCalledWith('Retry-After', '10'); + const body = res.json.mock.calls[0][0]; + expect(body.type).toBe('RATE_LIMIT_EXCEEDED'); + }); + + it('passes through unlimited when no wallet address is set (unauthenticated)', async () => { + const fakeRedis = buildFakeRedis(999); + mockGetRedis.mockReturnValue(fakeRedis); + const middleware = walletRateLimit({ + windowMs: 10_000, + max: 5, + keyPrefix: 'rl:test:', + }); + + const req = makeReq(undefined); + const res = makeRes(); + const next = jest.fn(); + await middleware(req, res, next); + + expect(next).toHaveBeenCalled(); + expect(fakeRedis.pipeline).not.toHaveBeenCalled(); + }); + + it('bypasses the limit for internal service calls with a matching key', async () => { + mockEnvConfig.INTERNAL_SERVICE_KEY = 'super-secret'; + const fakeRedis = buildFakeRedis(999); + mockGetRedis.mockReturnValue(fakeRedis); + const middleware = walletRateLimit({ + windowMs: 10_000, + max: 5, + keyPrefix: 'rl:test:', + }); + + const req = makeReq('GBUYER', { 'x-internal-service-key': 'super-secret' }); + const res = makeRes(); + const next = jest.fn(); + await middleware(req, res, next); + + expect(next).toHaveBeenCalled(); + expect(fakeRedis.pipeline).not.toHaveBeenCalled(); + }); + + it('does not bypass the limit when the internal service key header is wrong', async () => { + mockEnvConfig.INTERNAL_SERVICE_KEY = 'super-secret'; + const fakeRedis = buildFakeRedis(6); + mockGetRedis.mockReturnValue(fakeRedis); + const middleware = walletRateLimit({ + windowMs: 10_000, + max: 5, + keyPrefix: 'rl:test:', + }); + + const req = makeReq('GBUYER', { 'x-internal-service-key': 'wrong' }); + const res = makeRes(); + const next = jest.fn(); + await middleware(req, res, next); + + expect(next).not.toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(429); + }); + + it('fails open (allows the request) when Redis throws', async () => { + mockGetRedis.mockReturnValue({ + pipeline: () => ({ + zremrangebyscore: () => { + throw new Error('redis down'); + }, + }), + }); + const middleware = walletRateLimit({ + windowMs: 10_000, + max: 5, + keyPrefix: 'rl:test:', + }); + + const req = makeReq('GBUYER'); + const res = makeRes(); + const next = jest.fn(); + await middleware(req, res, next); + + expect(next).toHaveBeenCalled(); + expect(res.status).not.toHaveBeenCalled(); + }); +}); diff --git a/src/middlewares/wallet-rate-limit.middleware.ts b/src/middlewares/wallet-rate-limit.middleware.ts new file mode 100644 index 0000000..3c2a7ba --- /dev/null +++ b/src/middlewares/wallet-rate-limit.middleware.ts @@ -0,0 +1,123 @@ +// src/middlewares/wallet-rate-limit.middleware.ts +// Per-wallet sliding-window rate limiter backed by Redis. +// +// Unlike express-rate-limit's default in-memory store, this middleware keys +// on the authenticated wallet address (not IP) and uses a Redis sorted set +// per wallet so the window slides continuously rather than resetting on a +// fixed boundary. Intended for mutating, wallet-scoped endpoints such as the +// key purchase route, where a single wallet firing rapid-fire requests can +// front-run other buyers. + +import type { Response, NextFunction } from 'express'; +import { randomUUID } from 'crypto'; +import type { StellarSignedRequest } from './stellar-signature.middleware'; +import { getRedis } from '../utils/redis.utils'; +import { envConfig } from '../config'; +import { logger } from '../utils/logger.utils'; +import { sendRateLimitError } from '../utils/rate-limit-response.utils'; + +export interface WalletRateLimitOptions { + /** Sliding window duration in milliseconds. */ + windowMs: number; + /** Maximum number of requests allowed per wallet within the window. */ + max: number; + /** Redis key prefix, namespaced per route so limits don't collide. */ + keyPrefix: string; +} + +const INTERNAL_SERVICE_HEADER = 'x-internal-service-key'; + +function isInternalServiceCall(req: StellarSignedRequest): 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; +} + +/** + * Builds a sliding-window rate limit middleware scoped to the authenticated + * wallet address (`req.walletAddress`, set by `requireStellarSignature()`). + * + * Must be mounted after `requireStellarSignature()` so `walletAddress` is + * available. Requests without a resolved wallet address are passed through + * unlimited — signature verification is responsible for rejecting those. + * + * Fails open on Redis errors: a Redis outage logs a warning and allows the + * request through rather than blocking the purchase flow. + */ +export function walletRateLimit(options: WalletRateLimitOptions) { + const { windowMs, max, keyPrefix } = options; + + return async ( + req: StellarSignedRequest, + res: Response, + next: NextFunction + ): Promise => { + if (isInternalServiceCall(req)) { + next(); + return; + } + + const walletAddress = req.walletAddress; + if (!walletAddress) { + next(); + return; + } + + const redis = getRedis(); + const key = `${keyPrefix}${walletAddress}`; + const now = Date.now(); + const windowStart = now - windowMs; + const member = `${now}-${randomUUID()}`; + + try { + const pipeline = redis.pipeline(); + pipeline.zremrangebyscore(key, 0, windowStart); + pipeline.zadd(key, now, member); + pipeline.zcard(key); + pipeline.pexpire(key, windowMs); + const results = await pipeline.exec(); + + const countResult = results?.[2]; + const count = + countResult && !countResult[0] ? (countResult[1] as number) : 0; + + if (count > max) { + const retryAfterSeconds = Math.ceil(windowMs / 1000); + logger.warn( + { + type: 'rate_limit_breach', + walletAddress, + route: req.path, + limit: max, + windowMs, + timestamp: new Date(now).toISOString(), + }, + 'Wallet exceeded rate limit' + ); + sendRateLimitError(res, retryAfterSeconds); + return; + } + + next(); + } catch (error) { + logger.error( + { error, walletAddress, route: req.path }, + 'Rate limit check failed; allowing request through (fail open)' + ); + next(); + } + }; +} + +/** + * Rate limit applied to the key purchase (buy) endpoint: 5 requests per + * 10-second sliding window per wallet. + */ +export const buyKeyRateLimit = walletRateLimit({ + windowMs: 10_000, + max: 5, + keyPrefix: 'rl:buy:', +}); diff --git a/src/modules/auth/auth.controllers.ts b/src/modules/auth/auth.controllers.ts index 0ce19b4..9401dd2 100644 --- a/src/modules/auth/auth.controllers.ts +++ b/src/modules/auth/auth.controllers.ts @@ -1,4 +1,4 @@ -import { CreateUserWithPasswordSchema } from './auth.schemas'; +import { CreateUserWithPasswordType } from './auth.schemas'; import { AsyncController } from '../../types/auth.types'; import { checkUserEmailExists, createNewUserWithPassword } from './auth.utils'; import { SendMailAsync } from '../../utils/mail.utils'; @@ -12,7 +12,9 @@ export const httpRegisterUserWithPassword: AsyncController = async ( next ) => { try { - const validatedUserDetails = CreateUserWithPasswordSchema.parse(req.body); + // Body is already validated and stripped of unknown fields by the + // validateBody(CreateUserWithPasswordSchema) middleware on this route. + const validatedUserDetails = req.body as CreateUserWithPasswordType; const emailExists = await checkUserEmailExists( validatedUserDetails.email diff --git a/src/modules/auth/auth.routes.ts b/src/modules/auth/auth.routes.ts index e78c58f..c4f20e6 100644 --- a/src/modules/auth/auth.routes.ts +++ b/src/modules/auth/auth.routes.ts @@ -5,12 +5,18 @@ import { httpRefreshToken, } from './auth.controllers'; import { httpStellarChallenge } from './stellar-challenge.controller'; +import { validateBody } from '../../middlewares/validate-body.middleware'; +import { CreateUserWithPasswordSchema } from './auth.schemas'; const authRouter = Router(); authRouter.post('/challenge', httpStellarChallenge); authRouter.post('/login', httpLogin); -authRouter.post('/register', httpRegisterUserWithPassword); +authRouter.post( + '/register', + validateBody(CreateUserWithPasswordSchema), + httpRegisterUserWithPassword +); authRouter.post('/refresh', httpRefreshToken); export default authRouter; diff --git a/src/modules/creator/buy.controller.ts b/src/modules/creator/buy.controller.ts index 1709101..b1cedbd 100644 --- a/src/modules/creator/buy.controller.ts +++ b/src/modules/creator/buy.controller.ts @@ -2,38 +2,27 @@ import type { Response } from 'express'; import { z } from 'zod'; import type { StellarSignedRequest } from '../../middlewares/stellar-signature.middleware'; import { ErrorCode } from '../../constants/error.constants'; -import { - sendError, - sendSuccess, - zodIssuesToDetails, -} from '../../utils/api-response.utils'; +import { sendError, sendSuccess } from '../../utils/api-response.utils'; import { buyGateway } from './buy.service'; -const buySchema = z.object({ +export const buySchema = z.object({ quantity: z.number().int().positive(), key_cost_xlm: z.number().nonnegative(), fee_xlm: z.number().nonnegative().default(0), }); +export type BuyRequestBody = z.infer; + export async function httpBuyCreatorKey( req: StellarSignedRequest, res: Response ): Promise { - const parsed = buySchema.safeParse(req.body); - if (!parsed.success) { - sendError( - res, - 422, - ErrorCode.VALIDATION_ERROR, - 'Invalid buy request', - zodIssuesToDetails(parsed.error.issues) - ); - return; - } + // Body is already validated and stripped of unknown fields by the + // validateBody(buySchema) middleware on this route. + const body = req.body as BuyRequestBody; const walletAddress = req.walletAddress!; - const required = - parsed.data.key_cost_xlm * parsed.data.quantity + parsed.data.fee_xlm; + const required = body.key_cost_xlm * body.quantity + body.fee_xlm; const balance = await buyGateway.getXlmBalance(walletAddress); if (balance < required) { sendError( @@ -48,7 +37,7 @@ export async function httpBuyCreatorKey( const result = await buyGateway.submitBuy({ walletAddress, creatorId: String(req.params.id), - quantity: parsed.data.quantity, + quantity: body.quantity, }); sendSuccess(res, result, 200); } diff --git a/src/modules/creator/buy.integration.test.ts b/src/modules/creator/buy.integration.test.ts index 6dd263c..348daf4 100644 --- a/src/modules/creator/buy.integration.test.ts +++ b/src/modules/creator/buy.integration.test.ts @@ -3,14 +3,16 @@ import request from 'supertest'; import { Keypair } from '@stellar/stellar-base'; import { requireStellarSignature } from '../../middlewares/stellar-signature.middleware'; import { buildAuthHeaders } from '../../utils/test/auth-request.utils'; -import { httpBuyCreatorKey } from './buy.controller'; +import { httpBuyCreatorKey, buySchema } from './buy.controller'; import { buyGateway } from './buy.service'; +import { validateBody } from '../../middlewares/validate-body.middleware'; const app = express(); app.use(express.json()); app.post( '/api/v1/creators/:id/buy', requireStellarSignature(), + validateBody(buySchema), httpBuyCreatorKey ); diff --git a/src/modules/creator/creator-profile-update.integration.test.ts b/src/modules/creator/creator-profile-update.integration.test.ts index ef5172d..aca7c77 100644 --- a/src/modules/creator/creator-profile-update.integration.test.ts +++ b/src/modules/creator/creator-profile-update.integration.test.ts @@ -171,7 +171,10 @@ describe('PUT /api/v1/creators/:creatorId/profile — display name and bio persi ); }); - it('returns 400 when display name is too long', async () => { + // Status is 422 (not 400) because the request body fails schema + // validation via the shared validateBody middleware (#780) — 400 is + // reserved for malformed path params on this route. + it('returns 422 when display name is too long', async () => { const res = await supertest(app) .put(`/api/v1/creators/${TEST_CREATOR_ID}/profile`) .set('x-wallet-address', OWNER_WALLET_ADDRESS) @@ -180,7 +183,7 @@ describe('PUT /api/v1/creators/:creatorId/profile — display name and bio persi bio: 'Some bio', }); - expect(res.status).toBe(400); + expect(res.status).toBe(422); expect(res.body).toEqual( expect.objectContaining({ success: false, @@ -193,7 +196,7 @@ describe('PUT /api/v1/creators/:creatorId/profile — display name and bio persi expect(res.body.error.details[0].field).toBe('displayName'); }); - it('returns 400 when display name is empty string', async () => { + it('returns 422 when display name is empty string', async () => { const res = await supertest(app) .put(`/api/v1/creators/${TEST_CREATOR_ID}/profile`) .set('x-wallet-address', OWNER_WALLET_ADDRESS) @@ -202,7 +205,7 @@ describe('PUT /api/v1/creators/:creatorId/profile — display name and bio persi bio: 'Some bio', }); - expect(res.status).toBe(400); + expect(res.status).toBe(422); expect(res.body).toEqual( expect.objectContaining({ success: false, diff --git a/src/modules/creator/creator-profile.handlers.ts b/src/modules/creator/creator-profile.handlers.ts index aff633d..4000711 100644 --- a/src/modules/creator/creator-profile.handlers.ts +++ b/src/modules/creator/creator-profile.handlers.ts @@ -10,7 +10,7 @@ import { attachTimestampHeader } from '../../utils/timestamp-headers.utils'; import { logger } from '../../utils/logger.utils'; import { CreatorProfileParamsSchema, - UpsertCreatorProfileBodySchema, + UpsertCreatorProfileBody, } from './creator-profile.schemas'; import { getCreatorProfile, @@ -76,39 +76,13 @@ export async function upsertCreatorProfileHandler(req: Request, res: Response) { ); } - const bodyResult = UpsertCreatorProfileBodySchema.safeParse(req.body); - if (!bodyResult.success) { - // Log missing required fields with structured context - const missingFields = bodyResult.error.issues - .filter( - (issue: any) => - issue.code === 'invalid_type' && - issue.received === 'undefined' - ) - .map((issue: any) => issue.path.join('.')); - - if (missingFields.length > 0) { - logger.warn( - { - type: 'creator_profile_validation_error', - handler: 'upsertCreatorProfileHandler', - missingFields, - ...(req.requestId ? { requestId: req.requestId } : {}), - }, - 'Missing required fields in creator profile payload' - ); - } - - return sendValidationError( - res, - 'Invalid creator profile payload', - zodIssuesToDetails(bodyResult.error.issues) - ); - } + // Body is already validated and stripped of unknown fields by the + // validateBody(UpsertCreatorProfileBodySchema) middleware on this route. + const body = req.body as UpsertCreatorProfileBody; const profile = await upsertCreatorProfile( paramsResult.data.creatorId, - bodyResult.data + body ); return sendSuccess( res, diff --git a/src/modules/creator/creator.routes.ts b/src/modules/creator/creator.routes.ts index d370586..f0bd918 100644 --- a/src/modules/creator/creator.routes.ts +++ b/src/modules/creator/creator.routes.ts @@ -16,6 +16,8 @@ import { CREATOR_PUBLIC_ROUTE_NAMES } from '../../constants/creator-public-route import { requireCreatorProfileOwnership } from '../../middlewares/wallet-ownership.middleware'; import { validateCreatorParam } from '../../middlewares/creator-param.middleware'; import { requireStellarSignature } from '../../middlewares/stellar-signature.middleware'; +import { validateBody } from '../../middlewares/validate-body.middleware'; +import { UpsertCreatorProfileBodySchema } from './creator-profile.schemas'; const router = Router(); @@ -71,6 +73,7 @@ router.put( validateCreatorParam('creatorId'), requireStellarSignature(), requireCreatorProfileOwnership('creatorId'), + validateBody(UpsertCreatorProfileBodySchema), upsertCreatorProfileHandler ); // 405 handler for /:creatorId/profile diff --git a/src/modules/creator/post.controller.ts b/src/modules/creator/post.controller.ts index b34191f..597388a 100644 --- a/src/modules/creator/post.controller.ts +++ b/src/modules/creator/post.controller.ts @@ -3,16 +3,14 @@ import { z } from 'zod'; import type { StellarSignedRequest } from '../../middlewares/stellar-signature.middleware'; import { ErrorCode } from '../../constants/error.constants'; import { prisma } from '../../utils/prisma.utils'; -import { - sendError, - sendSuccess, - zodIssuesToDetails, -} from '../../utils/api-response.utils'; +import { sendError, sendSuccess } from '../../utils/api-response.utils'; -const postSchema = z.object({ +export const postSchema = z.object({ content: z.string().trim().min(1).max(5000), }); +export type CreatePostBody = z.infer; + function serializePost( post: { id: string; @@ -33,17 +31,9 @@ export async function httpCreatePost( req: StellarSignedRequest, res: Response ): Promise { - const parsed = postSchema.safeParse(req.body); - if (!parsed.success) { - sendError( - res, - 422, - ErrorCode.VALIDATION_ERROR, - 'Post content is required', - zodIssuesToDetails(parsed.error.issues) - ); - return; - } + // Body is already validated and stripped of unknown fields by the + // validateBody(postSchema) middleware on this route. + const body = req.body as CreatePostBody; const creatorId = String(req.params.id); const creator = await prisma.creatorProfile.findFirst({ @@ -63,7 +53,7 @@ export async function httpCreatePost( } const post = await prisma.creatorPost.create({ - data: { creatorId: creator.id, content: parsed.data.content }, + data: { creatorId: creator.id, content: body.content }, }); sendSuccess(res, serializePost(post, req.walletAddress!), 201); } diff --git a/src/modules/creator/post.integration.test.ts b/src/modules/creator/post.integration.test.ts index 901bb6d..283efee 100644 --- a/src/modules/creator/post.integration.test.ts +++ b/src/modules/creator/post.integration.test.ts @@ -13,13 +13,15 @@ jest.mock('../../utils/prisma.utils', () => ({ }, })); -import { httpCreatePost, httpListPosts } from './post.controller'; +import { httpCreatePost, httpListPosts, postSchema } from './post.controller'; +import { validateBody } from '../../middlewares/validate-body.middleware'; const app = express(); app.use(express.json()); app.post( '/api/v1/creators/:id/posts', requireStellarSignature(), + validateBody(postSchema), httpCreatePost ); app.get('/api/v1/creators/:id/posts', httpListPosts); diff --git a/src/modules/creators/creator-holders-cursor-pagination.integration.test.ts b/src/modules/creators/creator-holders-cursor-pagination.integration.test.ts new file mode 100644 index 0000000..443af8a --- /dev/null +++ b/src/modules/creators/creator-holders-cursor-pagination.integration.test.ts @@ -0,0 +1,169 @@ +// Integration test: cursor-based pagination on GET /creators/:id/holders (#778) +// +// Exercises the keyset pagination path added alongside the existing +// offset-based pagination: +// 1. ?cursor= resumes after the given holder and returns nextCursor +// 2. Last page returns hasMore=false and nextCursor=null +// 3. An invalid/tampered cursor returns 400 +// 4. A cursor for a holder that no longer exists returns 400 +// +// Uses Jest mocks — no database required. + +import { httpGetCreatorHolders } from './creator-holders.controller'; +import * as holdersService from './creator-holders.service'; +import { encodeHoldersCursor } from './creator-holders.service'; +import type { HolderRecord } from './creator-holders.service'; + +function makeReq( + params: Record = {}, + query: Record = {} +): any { + return { params, query }; +} + +function makeRes(): any { + const res: any = {}; + res.status = jest.fn().mockReturnValue(res); + res.json = jest.fn().mockReturnValue(res); + res.setHeader = jest.fn().mockReturnValue(res); + res.set = jest.fn().mockReturnValue(res); + return res; +} + +function makeNext(): jest.Mock { + return jest.fn(); +} + +// NOTE: parseCreatorId (src/utils/creator-id.utils.ts) currently only accepts +// positive-integer route params, so the id used here must be numeric — a +// pre-existing constraint unrelated to cursor pagination. +const CREATOR_STUB = { id: '12345', handle: 'alice' }; + +function makeHolder( + index: number, + overrides: Partial = {} +): HolderRecord { + const key_balance = (4 - index) * 10; + return { + wallet_address: `GWALLETADDRESS${String(index).padStart(46, '0')}`, + key_balance, + held_since: new Date(`2024-0${index}-01T00:00:00.000Z`), + key_count: key_balance, + share_percent: 0, + rank: index, + ...overrides, + }; +} + +describe('GET /creators/:id/holders — cursor pagination', () => { + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('resumes after the given cursor and returns a nextCursor when more holders exist', async () => { + jest + .spyOn(holdersService, 'findCreatorByIdOrHandle') + .mockResolvedValue(CREATOR_STUB); + + const nextHolder = makeHolder(2); + const fetchByCursorSpy = jest + .spyOn(holdersService, 'fetchCreatorHoldersByCursor') + .mockResolvedValue({ + holders: [nextHolder], + nextCursor: encodeHoldersCursor(nextHolder.wallet_address), + hasMore: true, + }); + + const cursor = encodeHoldersCursor(makeHolder(1).wallet_address); + const req = makeReq({ id: CREATOR_STUB.id }, { cursor, limit: '1' }); + const res = makeRes(); + await httpGetCreatorHolders(req, res, makeNext()); + + expect(res.status).toHaveBeenCalledWith(200); + const body = res.json.mock.calls[0][0]; + expect(body.data.items).toHaveLength(1); + expect(body.data.items[0].wallet_address).toBe(nextHolder.wallet_address); + expect(body.data.meta.hasMore).toBe(true); + expect(body.data.meta.nextCursor).toBeTruthy(); + expect(fetchByCursorSpy).toHaveBeenCalledWith( + CREATOR_STUB.id, + expect.objectContaining({ limit: 1 }), + makeHolder(1).wallet_address + ); + }); + + it('returns hasMore=false and nextCursor=null on the last page', async () => { + jest + .spyOn(holdersService, 'findCreatorByIdOrHandle') + .mockResolvedValue(CREATOR_STUB); + jest.spyOn(holdersService, 'fetchCreatorHoldersByCursor').mockResolvedValue({ + holders: [makeHolder(3)], + nextCursor: null, + hasMore: false, + }); + + const cursor = encodeHoldersCursor(makeHolder(2).wallet_address); + const req = makeReq({ id: CREATOR_STUB.id }, { cursor }); + const res = makeRes(); + await httpGetCreatorHolders(req, res, makeNext()); + + const body = res.json.mock.calls[0][0]; + expect(body.data.meta.hasMore).toBe(false); + expect(body.data.meta.nextCursor).toBeNull(); + }); + + it('returns 400 for a malformed cursor', async () => { + jest + .spyOn(holdersService, 'findCreatorByIdOrHandle') + .mockResolvedValue(CREATOR_STUB); + + const req = makeReq( + { id: CREATOR_STUB.id }, + { cursor: 'not-a-real-cursor' } + ); + const res = makeRes(); + await httpGetCreatorHolders(req, res, makeNext()); + + expect(res.status).toHaveBeenCalledWith(400); + const body = res.json.mock.calls[0][0]; + expect(body.success).toBe(false); + }); + + it('returns 400 when the cursor does not match a known holder', async () => { + jest + .spyOn(holdersService, 'findCreatorByIdOrHandle') + .mockResolvedValue(CREATOR_STUB); + jest + .spyOn(holdersService, 'fetchCreatorHoldersByCursor') + .mockResolvedValue(null); + + const cursor = encodeHoldersCursor('GNONEXISTENTWALLET'); + const req = makeReq({ id: CREATOR_STUB.id }, { cursor }); + const res = makeRes(); + await httpGetCreatorHolders(req, res, makeNext()); + + expect(res.status).toHaveBeenCalledWith(400); + }); + + it('offset-mode pagination is unaffected when no cursor is supplied', async () => { + jest + .spyOn(holdersService, 'findCreatorByIdOrHandle') + .mockResolvedValue(CREATOR_STUB); + const fetchOffsetSpy = jest + .spyOn(holdersService, 'fetchCreatorHolders') + .mockResolvedValue([[makeHolder(1)], 1]); + const fetchByCursorSpy = jest.spyOn( + holdersService, + 'fetchCreatorHoldersByCursor' + ); + + const req = makeReq({ id: CREATOR_STUB.id }, { limit: '20', offset: '0' }); + const res = makeRes(); + await httpGetCreatorHolders(req, res, makeNext()); + + expect(fetchOffsetSpy).toHaveBeenCalled(); + expect(fetchByCursorSpy).not.toHaveBeenCalled(); + const body = res.json.mock.calls[0][0]; + expect(body.data.meta.total).toBe(1); + }); +}); diff --git a/src/modules/creators/creator-holders.controller.ts b/src/modules/creators/creator-holders.controller.ts index 4b709cd..c5da74e 100644 --- a/src/modules/creators/creator-holders.controller.ts +++ b/src/modules/creators/creator-holders.controller.ts @@ -3,6 +3,8 @@ import { CreatorHoldersQuerySchema } from './creator-holders.schemas'; import { findCreatorByIdOrHandle, fetchCreatorHolders, + fetchCreatorHoldersByCursor, + decodeHoldersCursor, } from './creator-holders.service'; import { sendSuccess, @@ -49,6 +51,36 @@ export const httpGetCreatorHolders: AsyncController = async ( const creator = await findCreatorByIdOrHandle(String(creatorId)); if (!handleCreatorParamNotFound(res, creator)) return; + if (parsed.data.cursor) { + const decoded = decodeHoldersCursor(parsed.data.cursor); + if (!decoded.ok) { + return sendValidationError(res, 'Invalid pagination cursor', [ + { field: 'cursor', message: 'Cursor is malformed or has expired' }, + ]); + } + + const page = await fetchCreatorHoldersByCursor( + creator.id, + parsed.data, + decoded.ownerAddress + ); + if (!page) { + return sendValidationError(res, 'Invalid pagination cursor', [ + { field: 'cursor', message: 'Cursor does not match a known holder' }, + ]); + } + + attachTimestampHeader(res); + return sendSuccess(res, { + items: page.holders, + meta: { + limit: parsed.data.limit, + hasMore: page.hasMore, + nextCursor: page.nextCursor, + }, + }); + } + const [holders, total] = await fetchCreatorHolders( creator.id, parsed.data diff --git a/src/modules/creators/creator-holders.schemas.ts b/src/modules/creators/creator-holders.schemas.ts index 785a5b3..15ba38e 100644 --- a/src/modules/creators/creator-holders.schemas.ts +++ b/src/modules/creators/creator-holders.schemas.ts @@ -39,6 +39,12 @@ export const CreatorHoldersQuerySchema = z .enum(CREATOR_HOLDER_SORT_FIELDS) .optional() .default('key_balance'), + /** + * Opaque cursor produced by a previous page's `nextCursor` field. + * When provided, the endpoint switches to keyset (cursor-based) + * pagination and `offset` is ignored. + */ + cursor: z.string().min(1).optional(), }) .strict(); diff --git a/src/modules/creators/creator-holders.service.ts b/src/modules/creators/creator-holders.service.ts index 5feda42..86551ea 100644 --- a/src/modules/creators/creator-holders.service.ts +++ b/src/modules/creators/creator-holders.service.ts @@ -2,6 +2,7 @@ import { Prisma } from '@prisma/client'; import { prisma } from '../../utils/prisma.utils'; import { logger } from '../../utils/logger.utils'; import { CreatorHoldersQueryType } from './creator-holders.schemas'; +import { encodeCursor, decodeCursor, CursorChecksumError } from '../../utils/cursor.utils'; /** * Public-facing holder record returned by the holders endpoint. @@ -109,3 +110,135 @@ export async function fetchCreatorHolders( return [holders, total]; } + +/** Cursor payload for keyset-paginated holder pages. */ +interface HoldersCursorPayload { + ownerAddress: string; +} + +/** + * Encodes a holder's wallet address into an opaque, tamper-checked cursor + * string suitable for the `nextCursor` response field. + */ +export function encodeHoldersCursor(ownerAddress: string): string { + return encodeCursor({ ownerAddress }); +} + +export type DecodeHoldersCursorResult = + | { ok: true; ownerAddress: string } + | { ok: false }; + +/** + * Decodes and validates a client-supplied holders cursor. + * Returns `{ ok: false }` for malformed, tampered, or empty cursors. + */ +export function decodeHoldersCursor(raw: string): DecodeHoldersCursorResult { + try { + const payload = decodeCursor(raw); + if (typeof payload.ownerAddress !== 'string' || !payload.ownerAddress) { + return { ok: false }; + } + return { ok: true, ownerAddress: payload.ownerAddress }; + } catch (error) { + if (error instanceof CursorChecksumError) { + return { ok: false }; + } + return { ok: false }; + } +} + +export interface CursorHolderPage { + holders: HolderRecord[]; + nextCursor: string | null; + hasMore: boolean; +} + +/** + * Fetch a keyset-paginated page of key holders for a creator, resuming after + * the holder identified by `cursorOwnerAddress`. + * + * Uses the (ownerAddress, creatorId) unique index as the Prisma cursor so + * pagination stays stable and index-backed even as new holders are added. + * Over-fetches by one row to determine `hasMore` without a separate count + * query. + * + * @returns `null` if `cursorOwnerAddress` does not identify an existing + * holder row for this creator (i.e. the cursor is stale/invalid). + */ +export async function fetchCreatorHoldersByCursor( + creatorId: string, + query: CreatorHoldersQueryType, + cursorOwnerAddress: string +): Promise { + const { limit, sort } = query; + + const where: Prisma.KeyOwnershipWhereInput = { + creatorId, + balance: { gt: 0 }, + }; + + const orderBy: Prisma.KeyOwnershipOrderByWithRelationInput[] = + sort === 'held_since' + ? [{ createdAt: 'asc' }, { ownerAddress: 'asc' }] + : [{ balance: 'desc' }, { ownerAddress: 'asc' }]; + + const cursorRow = await prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: cursorOwnerAddress, + creatorId, + }, + }, + select: { id: true }, + }); + + if (!cursorRow) { + return null; + } + + const [rows, balanceSum] = await Promise.all([ + prisma.keyOwnership.findMany({ + where, + orderBy, + cursor: { + ownerAddress_creatorId: { + ownerAddress: cursorOwnerAddress, + creatorId, + }, + }, + skip: 1, + take: limit + 1, + select: { + ownerAddress: true, + balance: true, + createdAt: true, + }, + }), + prisma.keyOwnership.aggregate({ where, _sum: { balance: true } }), + ]); + + const totalKeys = Number(balanceSum._sum.balance ?? 0); + const hasMore = rows.length > limit; + const page = hasMore ? rows.slice(0, limit) : rows; + + const holders: HolderRecord[] = page.map((row, index) => { + const keyBalance = Number(row.balance); + return { + wallet_address: row.ownerAddress, + key_balance: keyBalance, + held_since: row.createdAt, + key_count: keyBalance, + share_percent: totalKeys > 0 ? (keyBalance / totalKeys) * 100 : 0, + // Position within this page only — cursor pagination doesn't track + // an absolute offset across pages. + rank: index + 1, + }; + }); + + const nextCursor = + hasMore && page.length > 0 + ? encodeHoldersCursor(page[page.length - 1].ownerAddress) + : null; + + return { holders, nextCursor, hasMore }; +} diff --git a/src/modules/creators/creator-leaderboard-volume.controller.test.ts b/src/modules/creators/creator-leaderboard-volume.controller.test.ts new file mode 100644 index 0000000..bb50b9f --- /dev/null +++ b/src/modules/creators/creator-leaderboard-volume.controller.test.ts @@ -0,0 +1,64 @@ +import { httpGetVolumeLeaderboard } from './creator-leaderboard-volume.controller'; +import * as service from './creator-leaderboard-volume.service'; +import type { VolumeLeaderboardEntry } from './creator-leaderboard-volume.service'; + +function makeRes(): any { + const res: any = {}; + res.status = jest.fn().mockReturnValue(res); + res.json = jest.fn().mockReturnValue(res); + res.setHeader = jest.fn().mockReturnValue(res); + res.set = jest.fn().mockReturnValue(res); + return res; +} + +const ENTRY: VolumeLeaderboardEntry = { + rank: 1, + keyId: 'creator-a', + creatorName: 'Alice', + avatarUrl: null, + totalVolume: '1000', + priceChange24h: 5, +}; + +describe('GET /api/v1/creators/leaderboard/volume', () => { + afterEach(() => jest.restoreAllMocks()); + + it('returns the leaderboard items wrapped in a success envelope', async () => { + jest.spyOn(service, 'getVolumeLeaderboard').mockResolvedValue([ENTRY]); + + const req: any = {}; + const res = makeRes(); + const next = jest.fn(); + await httpGetVolumeLeaderboard(req, res, next); + + expect(res.status).toHaveBeenCalledWith(200); + const body = res.json.mock.calls[0][0]; + expect(body.success).toBe(true); + expect(body.data.items).toEqual([ENTRY]); + }); + + it('returns an empty items array when there is no volume', async () => { + jest.spyOn(service, 'getVolumeLeaderboard').mockResolvedValue([]); + + const req: any = {}; + const res = makeRes(); + const next = jest.fn(); + await httpGetVolumeLeaderboard(req, res, next); + + const body = res.json.mock.calls[0][0]; + expect(body.data.items).toEqual([]); + }); + + it('forwards errors to next()', async () => { + const err = new Error('db down'); + jest.spyOn(service, 'getVolumeLeaderboard').mockRejectedValue(err); + + const req: any = {}; + const res = makeRes(); + const next = jest.fn(); + await httpGetVolumeLeaderboard(req, res, next); + + expect(next).toHaveBeenCalledWith(err); + expect(res.json).not.toHaveBeenCalled(); + }); +}); diff --git a/src/modules/creators/creator-leaderboard-volume.controller.ts b/src/modules/creators/creator-leaderboard-volume.controller.ts new file mode 100644 index 0000000..688e740 --- /dev/null +++ b/src/modules/creators/creator-leaderboard-volume.controller.ts @@ -0,0 +1,24 @@ +import { AsyncController } from '../../types/auth.types'; +import { getVolumeLeaderboard } from './creator-leaderboard-volume.service'; +import { sendSuccess } from '../../utils/api-response.utils'; +import { attachTimestampHeader } from '../../utils/timestamp-headers.utils'; + +/** + * Controller for GET /api/v1/creators/leaderboard/volume + * + * Returns the top 20 creator keys ranked by total trading volume (buys + + * sells) over a rolling window, cached in Redis for a short TTL. + */ +export const httpGetVolumeLeaderboard: AsyncController = async ( + _req, + res, + next +) => { + try { + const items = await getVolumeLeaderboard(); + attachTimestampHeader(res); + sendSuccess(res, { items }); + } catch (error) { + next(error); + } +}; diff --git a/src/modules/creators/creator-leaderboard-volume.service.test.ts b/src/modules/creators/creator-leaderboard-volume.service.test.ts new file mode 100644 index 0000000..c514608 --- /dev/null +++ b/src/modules/creators/creator-leaderboard-volume.service.test.ts @@ -0,0 +1,236 @@ +// Unit tests for the volume leaderboard (#785): +// - Aggregates buy + sell volume per creator from the Activity read model +// - Returns exactly 20 entries, sorted by descending volume +// - Serves from cache on repeated calls; falls back to live compute on a +// cache miss or Redis error +// - Computes priceChange24h from the creator's price snapshot +// - Returns an empty array when no trades exist + +const mockPrisma = { + activity: { findMany: jest.fn() }, + creatorProfile: { findMany: jest.fn() }, +}; +jest.mock('../../utils/prisma.utils', () => ({ prisma: mockPrisma })); + +jest.mock('../../utils/logger.utils', () => ({ + logger: { warn: jest.fn(), error: jest.fn(), debug: jest.fn(), info: jest.fn() }, +})); + +jest.mock('../../config', () => ({ + envConfig: { + LEADERBOARD_VOLUME_WINDOW_DAYS: 7, + LEADERBOARD_VOLUME_CACHE_TTL_SECONDS: 300, + }, +})); + +const mockRedisClient = { + get: jest.fn(), + set: jest.fn(), + del: jest.fn(), +}; +jest.mock('../../utils/redis.utils', () => ({ + getRedis: jest.fn(() => mockRedisClient), +})); + +import { + computeVolumeLeaderboard, + getVolumeLeaderboard, + invalidateVolumeLeaderboardCache, +} from './creator-leaderboard-volume.service'; + +function activity(creatorId: string, amount: number, priceAtTrade: string) { + return { + creatorId, + payload: { amount, price_at_trade: priceAtTrade }, + }; +} + +function creator( + id: string, + displayName: string, + avatarUrl: string | null, + currentPrice?: bigint, + price24hAgo?: bigint +) { + return { + id, + displayName, + avatarUrl, + priceSnapshot: + currentPrice !== undefined + ? { currentPrice, price24hAgo: price24hAgo ?? currentPrice } + : null, + }; +} + +describe('computeVolumeLeaderboard', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('returns an empty array when no trades exist', async () => { + mockPrisma.activity.findMany.mockResolvedValue([]); + + const result = await computeVolumeLeaderboard(); + + expect(result).toEqual([]); + expect(mockPrisma.creatorProfile.findMany).not.toHaveBeenCalled(); + }); + + it('sums buy and sell volume per creator (amount * price_at_trade)', async () => { + mockPrisma.activity.findMany.mockResolvedValue([ + activity('creator-a', 10, '100'), // 1000 + activity('creator-a', 5, '100'), // 500 (sell) -> total 1500 + ]); + mockPrisma.creatorProfile.findMany.mockResolvedValue([ + creator('creator-a', 'Alice', 'https://a.png', 100n, 90n), + ]); + + const result = await computeVolumeLeaderboard(); + + expect(result).toHaveLength(1); + expect(result[0]).toEqual( + expect.objectContaining({ + rank: 1, + keyId: 'creator-a', + creatorName: 'Alice', + avatarUrl: 'https://a.png', + totalVolume: '1500', + }) + ); + }); + + it('sorts entries by descending total volume', async () => { + mockPrisma.activity.findMany.mockResolvedValue([ + activity('low', 1, '100'), // 100 + activity('high', 100, '100'), // 10000 + ]); + mockPrisma.creatorProfile.findMany.mockResolvedValue([ + creator('low', 'Low Volume', null, 10n, 10n), + creator('high', 'High Volume', null, 10n, 10n), + ]); + + const result = await computeVolumeLeaderboard(); + + expect(result.map(entry => entry.keyId)).toEqual(['high', 'low']); + expect(result[0].rank).toBe(1); + expect(result[1].rank).toBe(2); + }); + + it('caps the result at 20 entries', async () => { + const activities = Array.from({ length: 25 }, (_, i) => + activity(`creator-${i}`, i + 1, '100') + ); + const creators = Array.from({ length: 25 }, (_, i) => + creator(`creator-${i}`, `Creator ${i}`, null, 10n, 10n) + ); + mockPrisma.activity.findMany.mockResolvedValue(activities); + mockPrisma.creatorProfile.findMany.mockResolvedValue(creators); + + const result = await computeVolumeLeaderboard(); + + expect(result).toHaveLength(20); + expect(result[0].rank).toBe(1); + expect(result[19].rank).toBe(20); + }); + + it('computes priceChange24h correctly from the price snapshot', async () => { + mockPrisma.activity.findMany.mockResolvedValue([ + activity('creator-a', 1, '100'), + ]); + mockPrisma.creatorProfile.findMany.mockResolvedValue([ + creator('creator-a', 'Alice', null, 120n, 100n), // +20% + ]); + + const result = await computeVolumeLeaderboard(); + + expect(result[0].priceChange24h).toBe(20); + }); + + it('returns priceChange24h=null when the creator has no price snapshot', async () => { + mockPrisma.activity.findMany.mockResolvedValue([ + activity('creator-a', 1, '100'), + ]); + mockPrisma.creatorProfile.findMany.mockResolvedValue([ + creator('creator-a', 'Alice', null), + ]); + + const result = await computeVolumeLeaderboard(); + + expect(result[0].priceChange24h).toBeNull(); + }); +}); + +describe('getVolumeLeaderboard (caching)', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('serves from Redis cache on a hit without querying the database', async () => { + const cachedPayload = [ + { + rank: 1, + keyId: 'creator-a', + creatorName: 'Alice', + avatarUrl: null, + totalVolume: '1000', + priceChange24h: 5, + }, + ]; + mockRedisClient.get.mockResolvedValue(JSON.stringify(cachedPayload)); + + const result = await getVolumeLeaderboard(); + + expect(result).toEqual(cachedPayload); + expect(mockPrisma.activity.findMany).not.toHaveBeenCalled(); + }); + + it('computes live and populates the cache on a miss', async () => { + mockRedisClient.get.mockResolvedValue(null); + mockPrisma.activity.findMany.mockResolvedValue([ + activity('creator-a', 1, '100'), + ]); + mockPrisma.creatorProfile.findMany.mockResolvedValue([ + creator('creator-a', 'Alice', null, 10n, 10n), + ]); + + const result = await getVolumeLeaderboard(); + + expect(result).toHaveLength(1); + expect(mockRedisClient.set).toHaveBeenCalledWith( + 'leaderboard:volume:v1', + expect.any(String), + 'EX', + 300 + ); + }); + + it('falls back to a live computation when Redis read fails', async () => { + mockRedisClient.get.mockRejectedValue(new Error('redis down')); + mockPrisma.activity.findMany.mockResolvedValue([]); + + const result = await getVolumeLeaderboard(); + + expect(result).toEqual([]); + }); +}); + +describe('invalidateVolumeLeaderboardCache', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('deletes the cache key', async () => { + mockRedisClient.del.mockResolvedValue(1); + + await invalidateVolumeLeaderboardCache(); + + expect(mockRedisClient.del).toHaveBeenCalledWith('leaderboard:volume:v1'); + }); + + it('does not throw when Redis is unavailable', async () => { + mockRedisClient.del.mockRejectedValue(new Error('redis down')); + + await expect(invalidateVolumeLeaderboardCache()).resolves.toBeUndefined(); + }); +}); diff --git a/src/modules/creators/creator-leaderboard-volume.service.ts b/src/modules/creators/creator-leaderboard-volume.service.ts new file mode 100644 index 0000000..8e3feb8 --- /dev/null +++ b/src/modules/creators/creator-leaderboard-volume.service.ts @@ -0,0 +1,217 @@ +// src/modules/creators/creator-leaderboard-volume.service.ts +// GET /api/v1/creators/leaderboard/volume (#785) +// +// Ranks creator keys by total trading volume (buys + sells combined) over a +// rolling window, backed by a short-lived Redis cache so the aggregation +// query doesn't run on every request. + +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { getRedis } from '../../utils/redis.utils'; +import { envConfig } from '../../config'; +import { compute24hPriceChange } from '../../utils/price.utils'; + +export interface VolumeLeaderboardEntry { + rank: number; + keyId: string; + creatorName: string; + avatarUrl: string | null; + /** Total trade volume in stroops over the window, as a decimal string. */ + totalVolume: string; + priceChange24h: number | null; +} + +const LEADERBOARD_LIMIT = 20; +const CACHE_KEY = 'leaderboard:volume:v1'; + +// The shared Redis client is configured with maxRetriesPerRequest: null +// (see src/utils/redis.utils.ts), so a command issued while Redis is +// unreachable would otherwise retry forever. Bound every cache operation +// here so a Redis outage degrades to "compute live" / "skip invalidation" +// instead of hanging the request (or, for invalidation, the indexer +// pipeline). +const REDIS_OP_TIMEOUT_MS = 1000; + +function withTimeout(promise: Promise, ms: number): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error(`Redis operation timed out after ${ms}ms`)), + ms + ); + promise.then( + value => { + clearTimeout(timer); + resolve(value); + }, + error => { + clearTimeout(timer); + reject(error); + } + ); + }); +} + +function windowMs(): number { + return envConfig.LEADERBOARD_VOLUME_WINDOW_DAYS * 24 * 60 * 60 * 1000; +} + +/** + * Recomputes the volume leaderboard directly from the Activity read model, + * bypassing the cache. Volume for a trade is `amount * price_at_trade` + * (both stored in the KEY_BOUGHT / KEY_SOLD activity payload), summed across + * both event types so buys and sells both count toward a key's volume. + */ +export async function computeVolumeLeaderboard(): Promise< + VolumeLeaderboardEntry[] +> { + const now = Date.now(); + const windowStart = new Date(now - windowMs()); + + const activities = await prisma.activity.findMany({ + where: { + type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] }, + creatorId: { not: null }, + createdAt: { gte: windowStart, lte: new Date(now) }, + }, + select: { creatorId: true, payload: true }, + }); + + const volumeByCreator = new Map(); + for (const activity of activities) { + if (!activity.creatorId) continue; + + const payload = activity.payload as Record; + if ( + payload && + payload.amount !== undefined && + payload.price_at_trade !== undefined && + payload.price_at_trade !== null + ) { + try { + const tradeVolume = + BigInt(Math.trunc(Number(payload.amount))) * + BigInt(payload.price_at_trade as string | number); + volumeByCreator.set( + activity.creatorId, + (volumeByCreator.get(activity.creatorId) ?? 0n) + tradeVolume + ); + } catch { + // Skip malformed payloads rather than fail the whole leaderboard. + continue; + } + } + } + + if (volumeByCreator.size === 0) { + return []; + } + + const creators = await prisma.creatorProfile.findMany({ + where: { id: { in: [...volumeByCreator.keys()] } }, + select: { + id: true, + displayName: true, + avatarUrl: true, + priceSnapshot: { + select: { currentPrice: true, price24hAgo: true }, + }, + }, + }); + + const unranked = creators.map(creator => { + const totalVolume = volumeByCreator.get(creator.id) ?? 0n; + const snapshot = creator.priceSnapshot; + const priceChange24h = snapshot + ? compute24hPriceChange(snapshot.currentPrice, snapshot.price24hAgo) + : null; + + return { + keyId: creator.id, + creatorName: creator.displayName, + avatarUrl: creator.avatarUrl, + totalVolume, + priceChange24h, + }; + }); + + unranked.sort((a, b) => { + if (a.totalVolume === b.totalVolume) { + return a.keyId < b.keyId ? -1 : a.keyId > b.keyId ? 1 : 0; + } + return a.totalVolume > b.totalVolume ? -1 : 1; + }); + + return unranked.slice(0, LEADERBOARD_LIMIT).map((entry, index) => ({ + rank: index + 1, + keyId: entry.keyId, + creatorName: entry.creatorName, + avatarUrl: entry.avatarUrl, + totalVolume: entry.totalVolume.toString(), + priceChange24h: entry.priceChange24h, + })); +} + +/** + * Returns the volume leaderboard, serving from the Redis cache when a fresh + * entry exists. Falls back to a live computation (without caching the + * result) if Redis is unreachable, so a cache outage never breaks the + * endpoint. + */ +export async function getVolumeLeaderboard(): Promise< + VolumeLeaderboardEntry[] +> { + const redis = getRedis(); + + try { + const cached = await withTimeout( + redis.get(CACHE_KEY), + REDIS_OP_TIMEOUT_MS + ); + if (cached) { + return JSON.parse(cached) as VolumeLeaderboardEntry[]; + } + } catch (error) { + logger.warn( + { error }, + 'Volume leaderboard cache read failed; computing live' + ); + } + + const leaderboard = await computeVolumeLeaderboard(); + + try { + await withTimeout( + redis.set( + CACHE_KEY, + JSON.stringify(leaderboard), + 'EX', + envConfig.LEADERBOARD_VOLUME_CACHE_TTL_SECONDS + ), + REDIS_OP_TIMEOUT_MS + ); + } catch (error) { + logger.warn( + { error }, + 'Volume leaderboard cache write failed; serving uncached result' + ); + } + + return leaderboard; +} + +/** + * Invalidates the cached volume leaderboard. Called after a new trade is + * recorded so the leaderboard reflects it within the next request instead of + * waiting out the full TTL. + */ +export async function invalidateVolumeLeaderboardCache(): Promise { + try { + const redis = getRedis(); + await withTimeout(redis.del(CACHE_KEY), REDIS_OP_TIMEOUT_MS); + } catch (error) { + logger.warn( + { error }, + 'Failed to invalidate volume leaderboard cache' + ); + } +} diff --git a/src/modules/creators/creators.routes.ts b/src/modules/creators/creators.routes.ts index d16475c..6fcf082 100644 --- a/src/modules/creators/creators.routes.ts +++ b/src/modules/creators/creators.routes.ts @@ -8,6 +8,7 @@ import { httpGetCreatorAnalytics, } from './creators.controllers'; import { httpGetCreatorHolders } from './creator-holders.controller'; +import { httpGetVolumeLeaderboard } from './creator-leaderboard-volume.controller'; import { cacheControl } from '../../middlewares/cache-control.middleware'; import { CREATOR_PUBLIC_ROUTE_CACHE_PRESETS } from '../../constants/creator-public-cache.constants'; import { CREATOR_PUBLIC_ROUTE_NAMES } from '../../constants/creator-public-routes.constants'; @@ -18,8 +19,14 @@ import { requireCreatorProfileOwnership, } from '../../middlewares/wallet-ownership.middleware'; import { requireStellarSignature } from '../../middlewares/stellar-signature.middleware'; -import { httpBuyCreatorKey } from '../creator/buy.controller'; -import { httpCreatePost, httpListPosts } from '../creator/post.controller'; +import { buyKeyRateLimit } from '../../middlewares/wallet-rate-limit.middleware'; +import { validateBody } from '../../middlewares/validate-body.middleware'; +import { httpBuyCreatorKey, buySchema } from '../creator/buy.controller'; +import { + httpCreatePost, + httpListPosts, + postSchema, +} from '../creator/post.controller'; const creatorsRouter = Router(); @@ -32,6 +39,8 @@ creatorsRouter.post( '/:id/buy', validateCreatorParam('id'), requireStellarSignature(), + buyKeyRateLimit, + validateBody(buySchema), httpBuyCreatorKey ); creatorsRouter.get('/:id/posts', validateCreatorParam('id'), httpListPosts); @@ -39,6 +48,7 @@ creatorsRouter.post( '/:id/posts', validateCreatorParam('id'), requireStellarSignature(), + validateBody(postSchema), httpCreatePost ); @@ -142,6 +152,20 @@ creatorsRouter.get( httpGetCreatorLeaderboard ); +/** + * GET /api/v1/creators/leaderboard/volume + * + * Top 20 creator keys ranked by total trading volume (buys + sells) over a + * rolling window (default 7 days, LEADERBOARD_VOLUME_WINDOW_DAYS). Cached in + * Redis for LEADERBOARD_VOLUME_CACHE_TTL_SECONDS (default 5 minutes) and + * invalidated whenever a new trade is indexed. + */ +creatorsRouter.get( + '/leaderboard/volume', + createCreatorReadMetricsMiddleware('list'), + httpGetVolumeLeaderboard +); + /** * GET /api/v1/creators/:id * diff --git a/src/modules/indexer/indexer-pipeline.integration.test.ts b/src/modules/indexer/indexer-pipeline.integration.test.ts index 975dcda..2850b92 100644 --- a/src/modules/indexer/indexer-pipeline.integration.test.ts +++ b/src/modules/indexer/indexer-pipeline.integration.test.ts @@ -32,6 +32,15 @@ jest.mock('../../utils/logger.utils', () => ({ }, })); +// processTradeEvents invalidates the volume leaderboard cache (#785) after +// creating each Activity row — stub Redis so that call resolves immediately +// instead of attempting a real connection. +jest.mock('../../utils/redis.utils', () => ({ + getRedis: jest.fn(() => ({ + del: jest.fn().mockResolvedValue(1), + })), +})); + describe('processTradeEvents integration test', () => { const mockPrisma = prisma as unknown as { activity: { create: jest.Mock }; diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 06f3027..1159a75 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/src/modules/indexer/indexer-pipeline.service.ts @@ -7,6 +7,7 @@ import { logger } from '../../utils/logger.utils'; import { processIndexerChainEvents, IndexerChainEvent } from '../../utils/indexer-event-processor.utils'; import { dedupeChainEvents } from '../../utils/indexer-dedupe.utils'; import { logSellTransactionConfirmed } from '../../utils/sell-transaction-logger.utils'; +import { invalidateVolumeLeaderboardCache } from '../creators/creator-leaderboard-volume.service'; /** * Processes a batch of on-chain trade events (KEY_BOUGHT or KEY_SOLD). @@ -55,6 +56,10 @@ export async function processTradeEvents(events: IndexerChainEvent[]): Promise