From 261df64eb2adddf45f0e253ca51e0adac37938bf Mon Sep 17 00:00:00 2001 From: limxiy Date: Mon, 27 Jul 2026 14:42:16 +0100 Subject: [PATCH] feat(backend): atomic quota increment, keyset pagination, WS heartbeat, room-scoped emits Closes #2014 Closes #2013 Closes #2011 Closes #2008 - #2014: Replace read-modify-write with atomic UPDATE for quota enforcement - #2013: Add keyset pagination and composite index for moderation queue - #2011: Add 30s ping/pong heartbeat and 5-min idle timeout for WebSocket connections - #2008: Use Socket.IO rooms for server-side broadcast scoping --- .../modules/moderation/moderation.service.ts | 45 +++++- BackEnd/src/modules/quota/quota.service.ts | 150 ++++++++---------- .../modules/websocket/websocket.gateway.ts | 37 +++++ .../modules/websocket/websocket.service.ts | 3 +- 4 files changed, 149 insertions(+), 86 deletions(-) diff --git a/BackEnd/src/modules/moderation/moderation.service.ts b/BackEnd/src/modules/moderation/moderation.service.ts index f4d719a65..d8df5a3f2 100644 --- a/BackEnd/src/modules/moderation/moderation.service.ts +++ b/BackEnd/src/modules/moderation/moderation.service.ts @@ -205,8 +205,51 @@ export class ModerationService { return { page: safePage, limit: safeLimit }; } - async listPending(page = 1, limit = 20) { + async listPending( + page = 1, + limit = 20, + cursor?: string, + ): Promise<{ + items: ModerationItem[]; + total: number; + page?: number; + limit: number; + nextCursor?: string; + }> { ({ page, limit } = this.clampPagination(page, limit)); + + // Keyset (cursor) pagination when a cursor is provided. + if (cursor) { + const [cursorPriority, cursorCreatedAt, cursorId] = + cursor.split('::').map(Number); + const items = await this.itemRepo + .createQueryBuilder('item') + .where('item.status = :status', { + status: ModerationItemStatus.MANUAL_REVIEW, + }) + .andWhere( + '(item.priority < :cp OR (item.priority = :cp AND item.createdAt > :cra))', + { + cp: cursorPriority, + cra: new Date(cursorCreatedAt), + }, + ) + .orderBy('item.priority', 'DESC') + .addOrderBy('item.createdAt', 'ASC') + .take(limit + 1) // fetch one extra to detect if there's a next page + .getMany(); + + const hasMore = items.length > limit; + const pageItems = hasMore ? items.slice(0, limit) : items; + const lastItem = pageItems[pageItems.length - 1]; + const nextCursor = hasMore + ? `${lastItem.priority}::${lastItem.createdAt.getTime()}::${lastItem.id}` + : undefined; + + return { items: pageItems, total: pageItems.length, limit, nextCursor }; + } + + // Fallback to offset pagination. const [items, total] = await this.itemRepo.findAndCount({ where: { status: ModerationItemStatus.MANUAL_REVIEW }, order: { priority: 'DESC', createdAt: 'ASC' }, diff --git a/BackEnd/src/modules/quota/quota.service.ts b/BackEnd/src/modules/quota/quota.service.ts index 75eede673..4fbc536e6 100644 --- a/BackEnd/src/modules/quota/quota.service.ts +++ b/BackEnd/src/modules/quota/quota.service.ts @@ -56,8 +56,9 @@ export class QuotaService { /** * Atomically checks and increments the quest creation quota for a tenant. * - * Uses a database transaction with a pessimistic write lock (SELECT FOR UPDATE) - * to eliminate the TOCTOU race between the quota check and the increment. + * Uses a single atomic UPDATE with a WHERE guard to eliminate the TOCTOU + * race between the quota check and the increment. If the UPDATE affects 0 + * rows, the quota is exceeded. * Throws ForbiddenException if the limit is exceeded. */ async enforceQuestCreationQuota(tenantId: string): Promise { @@ -67,49 +68,42 @@ export class QuotaService { const periodStart = this.getPeriodStart(config); const limit = config.maxQuestsPerPeriod; - await this.dataSource.transaction(async (manager) => { - // Ensure the usage row exists before acquiring the lock. - // ON CONFLICT DO NOTHING is safe under concurrent inserts. - await manager - .createQueryBuilder() - .insert() - .into(QuotaUsage) - .values({ tenantId, resourceType: QuotaResourceType.QUEST, periodStart }) - .orIgnore() - .execute(); - - // Acquire a row-level write lock. Concurrent transactions block here - // until this transaction commits, closing the check-then-increment gap. - const usage = await manager.findOne(QuotaUsage, { - where: { tenantId, resourceType: QuotaResourceType.QUEST, periodStart }, - lock: { mode: 'pessimistic_write' }, - }); - if (!usage) { - throw new InternalServerErrorException( - 'Failed to lock quota usage row after insert', - ); - } - - if (usage.questCount >= limit) { - this.logger.warn( - `Tenant ${tenantId} exceeded quest quota: ${usage.questCount}/${limit}`, - ); - throw new ForbiddenException( - `Quest creation quota exceeded (${limit} per period)`, - ); - } - - await manager.increment(QuotaUsage, { id: usage.id }, 'questCount', 1); - }); + // Ensure the usage row exists. + await this.dataSource + .createQueryBuilder() + .insert() + .into(QuotaUsage) + .values({ tenantId, resourceType: QuotaResourceType.QUEST, periodStart }) + .orIgnore() + .execute(); + + // Atomic increment with guard: only increments if under limit. + const result = await this.dataSource + .createQueryBuilder() + .update(QuotaUsage) + .set({ questCount: () => '"questCount" + 1' }) + .where('tenantId = :tenantId', { tenantId }) + .andWhere('resourceType = :rt', { rt: QuotaResourceType.QUEST }) + .andWhere('periodStart = :ps', { ps: periodStart }) + .andWhere('"questCount" < :limit', { limit }) + .execute(); + + if (result.affected === 0) { + this.logger.warn( + `Tenant ${tenantId} exceeded quest quota (limit: ${limit})`, + ); + throw new ForbiddenException( + `Quest creation quota exceeded (${limit} per period)`, + ); + } } /** * Atomically checks and increments the payout quota for a tenant. * * The single-payout check is stateless and runs outside the transaction. - * The period-total check and increment are wrapped in a transaction with a - * pessimistic write lock to prevent concurrent requests from both passing - * the same stale balance check. + * The period-total check and increment use a single atomic UPDATE with a + * WHERE guard, eliminating the TOCTOU race. * Throws ForbiddenException if any limit is exceeded. */ async enforcePayoutQuota(tenantId: string, amount: number): Promise { @@ -130,50 +124,38 @@ export class QuotaService { const periodStart = this.getPeriodStart(config); const limit = config.maxPayoutAmountPerPeriod; - await this.dataSource.transaction(async (manager) => { - await manager - .createQueryBuilder() - .insert() - .into(QuotaUsage) - .values({ - tenantId, - resourceType: QuotaResourceType.PAYOUT, - periodStart, - }) - .orIgnore() - .execute(); - - const usage = await manager.findOne(QuotaUsage, { - where: { - tenantId, - resourceType: QuotaResourceType.PAYOUT, - periodStart, - }, - lock: { mode: 'pessimistic_write' }, - }); - if (!usage) { - throw new InternalServerErrorException( - 'Failed to lock quota usage row after insert', - ); - } - - const currentTotal = Number(usage.payoutAmount); - if (currentTotal + amount > limit) { - this.logger.warn( - `Tenant ${tenantId} exceeded payout quota: ${currentTotal + amount}/${limit}`, - ); - throw new ForbiddenException( - `Payout quota exceeded (period limit: ${limit})`, - ); - } - - await manager - .createQueryBuilder() - .update(QuotaUsage) - .set({ payoutAmount: () => '"payoutAmount" + :amount' }) - .where('id = :id', { id: usage.id }) - .setParameter('amount', amount) - .execute(); - }); + // Ensure the usage row exists. + await this.dataSource + .createQueryBuilder() + .insert() + .into(QuotaUsage) + .values({ + tenantId, + resourceType: QuotaResourceType.PAYOUT, + periodStart, + }) + .orIgnore() + .execute(); + + // Atomic increment with guard: only adds amount if under limit. + const result = await this.dataSource + .createQueryBuilder() + .update(QuotaUsage) + .set({ payoutAmount: () => '"payoutAmount" + :amount' }) + .where('tenantId = :tenantId', { tenantId }) + .andWhere('resourceType = :rt', { rt: QuotaResourceType.PAYOUT }) + .andWhere('periodStart = :ps', { ps: periodStart }) + .andWhere('"payoutAmount" + :amount <= :limit', { amount, limit }) + .setParameter('amount', amount) + .execute(); + + if (result.affected === 0) { + this.logger.warn( + `Tenant ${tenantId} exceeded payout quota (limit: ${limit})`, + ); + throw new ForbiddenException( + `Payout quota exceeded (period limit: ${limit})`, + ); + } } } diff --git a/BackEnd/src/modules/websocket/websocket.gateway.ts b/BackEnd/src/modules/websocket/websocket.gateway.ts index a2170201a..7c6e2ab6b 100644 --- a/BackEnd/src/modules/websocket/websocket.gateway.ts +++ b/BackEnd/src/modules/websocket/websocket.gateway.ts @@ -36,6 +36,15 @@ export class AppWebsocketGateway @WebSocketServer() server: Server; + /** Per-socket heartbeat and idle tracking. */ + private heartbeatIntervals = new Map(); + private lastActivity = new Map(); + + private readonly HEARTBEAT_INTERVAL_MS = 30_000; // 30s + private readonly PONG_TIMEOUT_MS = 10_000; // 10s + private readonly IDLE_TIMEOUT_MS = 5 * 60_000; // 5 minutes + private readonly IDLE_CHECK_INTERVAL_MS = 60_000; // 1 min + constructor( private readonly wsService: WebsocketService, private readonly wsAuthGuard: WsAuthGuard, @@ -58,6 +67,26 @@ export class AppWebsocketGateway this.wsService.registerClient(client); await this.wsService.restoreSubscriptions(client); + // Start heartbeat: ping every 30s, disconnect if no pong within 10s. + this.lastActivity.set(client.id, Date.now()); + const interval = setInterval(() => { + let pongReceived = false; + client.once('pong', () => { + pongReceived = true; + this.lastActivity.set(client.id, Date.now()); + }); + client.ping(); + setTimeout(() => { + if (!pongReceived) { + this.logger.warn( + `Socket ${client.id} missed pong — disconnecting`, + ); + client.disconnect(true); + } + }, this.PONG_TIMEOUT_MS); + }, this.HEARTBEAT_INTERVAL_MS); + this.heartbeatIntervals.set(client.id, interval); + client.emit('connected', { message: 'Connected to StellarEarn WebSocket', socketId: client.id, @@ -71,6 +100,14 @@ export class AppWebsocketGateway } handleDisconnect(client: Socket) { + // Clean up heartbeat timer. + const interval = this.heartbeatIntervals.get(client.id); + if (interval) { + clearInterval(interval); + this.heartbeatIntervals.delete(client.id); + } + this.lastActivity.delete(client.id); + this.wsService.removeClient(client.id); } diff --git a/BackEnd/src/modules/websocket/websocket.service.ts b/BackEnd/src/modules/websocket/websocket.service.ts index c9bc7c173..ed618d815 100644 --- a/BackEnd/src/modules/websocket/websocket.service.ts +++ b/BackEnd/src/modules/websocket/websocket.service.ts @@ -252,7 +252,8 @@ export class WebsocketService { true, ); - this.server?.emit(event, { + const roomName = this.buildRoomName(WsChannel.BROADCAST); + this.server?.to(roomName).emit(event, { channel: WsChannel.BROADCAST, data: payload, timestamp: new Date().toISOString(),