feat(backend): atomic quota increment, keyset pagination, WS heartbeat, room-scoped emits

BackEnd/src/modules/moderation/moderation.service.ts

@@ -201,8 +201,51 @@
     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' },

BackEnd/src/modules/quota/quota.service.ts

@@ -2,7 +2,7 @@
   Injectable,
   Logger,
   ForbiddenException,
   InternalServerErrorException,
 } from '@nestjs/common';
 import { InjectRepository, InjectDataSource } from '@nestjs/typeorm';
 import { Repository, DataSource } from 'typeorm';
@@ -59,8 +59,9 @@
   /**
    * 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<void> {
@@ -70,44 +71,33 @@
     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);
-    });
+    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)`,
+      );
+    }
 
     await this.updateCachedQuotaUsage(
       tenantId,
@@ -120,9 +110,8 @@
    * 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<void> {
@@ -143,51 +132,39 @@
     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::text = :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})`,
+      );
+    }
 
     await this.updateCachedQuotaUsage(
       tenantId,

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<string, NodeJS.Timeout>();
+  private lastActivity = new Map<string, number>();
+
+  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);
   }
 

BackEnd/src/modules/websocket/websocket.service.ts

@@ -297,7 +297,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(),