feat: batch payout txs, background analytics, submissions pagination

BackEnd/src/modules/analytics/CHANGELOG.md

@@ -5,3 +5,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 and this module adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
+
+### Added
+- Background platform analytics computation via scheduled cron job (every 5 minutes)
+- Snapshot-based serving: `getPlatformStats()` reads pre-computed `AnalyticsSnapshot` data instead of running heavy synchronous queries
+- Automatic fallback to live computation when no fresh snapshot exists
+- `computeAndStorePlatformStats()` persists computed stats to `analytics_snapshots` table
+- Metrics tracking: `analytics_computation_total` (source: snapshot|live) and `analytics_computation_duration_seconds` histogram
+- Background cron job `computePlatformAnalytics()` on `EVERY_5_MINUTES` schedule

BackEnd/src/modules/analytics/services/platform-analytics.service.ts

@@ -1,6 +1,7 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
-import { Repository } from 'typeorm';
+import { Cron, CronExpression } from '@nestjs/schedule';
+import { Repository, MoreThanOrEqual } from 'typeorm';
 import { Quest } from '../entities/quest.entity';
 import { Submission, SubmissionStatus } from '../entities/submission.entity';
 import { Payout } from '../entities/payout.entity';
@@ -13,9 +14,13 @@ import { DateRangeUtil } from '../utils/date-range.util';
 import { ConversionUtil } from '../utils/conversion.util';
 import { CacheService } from './cache.service';
 import { User as AnalyticsUser } from '../entities/user.entity';
+import { AnalyticsSnapshot, SnapshotType } from '../entities/analytics-snapshot.entity';
+import { MetricsService } from '../../../common/services/metrics.service';
 
 @Injectable()
 export class PlatformAnalyticsService {
+  private readonly logger = new Logger(PlatformAnalyticsService.name);
+
   constructor(
     @InjectRepository(AnalyticsUser)
     private userRepository: Repository<AnalyticsUser>,
@@ -25,86 +30,123 @@ export class PlatformAnalyticsService {
     private submissionRepository: Repository<Submission>,
     @InjectRepository(Payout)
     private payoutRepository: Repository<Payout>,
+    @InjectRepository(AnalyticsSnapshot)
+    private snapshotRepository: Repository<AnalyticsSnapshot>,
     private cacheService: CacheService,
+    private metricsService: MetricsService,
   ) {}
 
-  /**
-   * Get platform-wide statistics
-   */
   async getPlatformStats(query: AnalyticsQueryDto): Promise<PlatformStatsDto> {
     const { startDate, endDate } = DateRangeUtil.parseDateRange(
       query.startDate,
       query.endDate,
     );
     DateRangeUtil.validateMaxRange(startDate, endDate);
 
-    const cacheKey = this.cacheService.generateKey('platform', {
-      start: startDate.toISOString(),
-      end: endDate.toISOString(),
-      granularity: query.granularity,
+    const snapshot = await this.snapshotRepository.findOne({
+      where: {
+        type: SnapshotType.PLATFORM,
+        date: MoreThanOrEqual(new Date(Date.now() - 5 * 60 * 1000)),
+      },
+      order: { date: 'DESC' },
     });
 
-    return this.cacheService.wrap(
-      cacheKey,
-      async () => {
-        const [
-          totalUsers,
-          totalQuests,
-          totalSubmissions,
-          approvedSubmissions,
-          totalPayouts,
-          totalRewardsDistributed,
-          activeUsers,
-          questsByStatus,
-          submissionsByStatus,
-          allSubmissions,
-          timeSeries,
-        ] = await Promise.all([
-          this.getTotalUsers(startDate, endDate),
-          this.getTotalQuests(startDate, endDate),
-          this.getTotalSubmissions(startDate, endDate),
-          this.getApprovedSubmissions(startDate, endDate),
-          this.getTotalPayouts(startDate, endDate),
-          this.getTotalRewardsDistributed(startDate, endDate),
-          this.getActiveUsers(startDate, endDate),
-          this.getQuestsByStatus(startDate, endDate),
-          this.getSubmissionsByStatus(startDate, endDate),
-          this.getAllSubmissions(startDate, endDate),
-          this.getTimeSeries(
-            startDate,
-            endDate,
-            query.granularity || Granularity.DAY,
-          ),
-        ]);
-
-        const approvalRate = ConversionUtil.calculateApprovalRate(
-          approvedSubmissions,
-          totalSubmissions,
-        );
-
-        const avgApprovalTime = ConversionUtil.calculateAverageTime(
-          allSubmissions.filter((s) => s.status === SubmissionStatus.APPROVED),
-          'submittedAt', // Using submittedAt
-          'reviewedAt', // Using reviewedAt
-        );
-
-        return {
-          totalUsers,
-          totalQuests,
-          totalSubmissions,
-          approvedSubmissions,
-          totalPayouts,
-          totalRewardsDistributed,
-          approvalRate,
-          avgApprovalTime,
-          activeUsers,
-          timeSeries,
-          questsByStatus,
-          submissionsByStatus,
-        };
+    if (snapshot) {
+      this.metricsService.incrementCounter('analytics_computation_total', { source: 'snapshot' });
+      return snapshot.metrics as unknown as PlatformStatsDto;
+    }
+
+    this.metricsService.incrementCounter('analytics_computation_total', { source: 'live' });
+    return this.computeAndStorePlatformStats(startDate, endDate, query.granularity || Granularity.DAY);
+  }
+
+  async computeAndStorePlatformStats(
+    startDate: Date,
+    endDate: Date,
+    granularity: Granularity,
+  ): Promise<PlatformStatsDto> {
+    const startTime = Date.now();
+
+    const [
+      totalUsers,
+      totalQuests,
+      totalSubmissions,
+      approvedSubmissions,
+      totalPayouts,
+      totalRewardsDistributed,
+      activeUsers,
+      questsByStatus,
+      submissionsByStatus,
+      allSubmissions,
+      timeSeries,
+    ] = await Promise.all([
+      this.getTotalUsers(startDate, endDate),
+      this.getTotalQuests(startDate, endDate),
+      this.getTotalSubmissions(startDate, endDate),
+      this.getApprovedSubmissions(startDate, endDate),
+      this.getTotalPayouts(startDate, endDate),
+      this.getTotalRewardsDistributed(startDate, endDate),
+      this.getActiveUsers(startDate, endDate),
+      this.getQuestsByStatus(startDate, endDate),
+      this.getSubmissionsByStatus(startDate, endDate),
+      this.getAllSubmissions(startDate, endDate),
+      this.getTimeSeries(startDate, endDate, granularity),
+    ]);
+
+    const approvalRate = ConversionUtil.calculateApprovalRate(
+      approvedSubmissions,
+      totalSubmissions,
+    );
+
+    const avgApprovalTime = ConversionUtil.calculateAverageTime(
+      allSubmissions.filter((s) => s.status === SubmissionStatus.APPROVED),
+      'submittedAt',
+      'reviewedAt',
+    );
+
+    const stats: PlatformStatsDto = {
+      totalUsers,
+      totalQuests,
+      totalSubmissions,
+      approvedSubmissions,
+      totalPayouts,
+      totalRewardsDistributed,
+      approvalRate,
+      avgApprovalTime,
+      activeUsers,
+      timeSeries,
+      questsByStatus,
+      submissionsByStatus,
+    };
+
+    await this.snapshotRepository.upsert(
+      {
+        type: SnapshotType.PLATFORM,
+        date: new Date(),
+        metrics: stats as unknown as Record<string, any>,
       },
-      300, // 5 minutes TTL
+      ['type', 'date'],
     );
+
+    this.metricsService.observeHistogram(
+      'analytics_computation_duration_seconds',
+      (Date.now() - startTime) / 1000,
+    );
+
+    return stats;
+  }
+
+  @Cron(CronExpression.EVERY_5_MINUTES)
+  async computePlatformAnalytics(): Promise<void> {
+    const now = new Date();
+    const startDate = new Date(now.getFullYear(), now.getMonth(), now.getDate() - 30);
+    const endDate = now;
+    try {
+      await this.computeAndStorePlatformStats(startDate, endDate, Granularity.DAY);
+      this.logger.log('Background platform analytics computation completed');
+    } catch (error) {
+      this.logger.error('Background platform analytics computation failed', error);
+    }
   }
 
   private async getTotalUsers(startDate: Date, endDate: Date): Promise<number> {

BackEnd/src/modules/payouts/CHANGELOG.md

@@ -9,6 +9,12 @@ and this module adheres to [Semantic Versioning](https://semver.org/).
 ### Added
 
 - Partial indexes (`WHERE "deletedAt" IS NULL`) on `Payout` for `status` and `[type, status]` columns to speed up active-payout queries (#2000).
+- `processBatchPayouts()` method that groups PENDING/RETRY_SCHEDULED payouts by asset and submits them in batches of up to 100 operations per Stellar transaction via `StellarService.sendBatchPayments()` (#1981).
+- `processPendingBatch()` cron job (`EVERY_30_SECONDS`) that drives the batch payout processing loop (#1981).
+- `claimPayout` now sets status to `PENDING` instead of `PROCESSING` so the batch cron picks up the payout (#1981).
+- `executeStellarPayment` now delegates to `stellarService.sendPayment()` for production payments instead of throwing (#1981).
+- `StellarModule` imported into `PayoutsModule` so `StellarService` is available for injection (#1981).
+- Batch payout metrics (`batch_payout_total`, `batch_payout_operations`, `batch_payout_size`) recorded via `MetricsService` (#1981).
 - Redis-backed payout status polling cache via `JobResultStatusCacheService` to avoid Postgres reads on repeated `GET /payouts/:id` polls (#1983).
 
 ### Changed

BackEnd/src/modules/payouts/payouts.module.ts

@@ -12,6 +12,7 @@ import { FraudRiskRulesService } from './services/fraud-risk-rules.service';
 import { QuotaModule } from '../quota/quota.module';
 import { JobsModule } from '../jobs/jobs.module';
 import { BulkheadService } from '../../common/services/bulkhead.service';
+import { StellarModule } from '../stellar/stellar.module';
 
 @Module({
   imports: [
@@ -20,6 +21,7 @@ import { BulkheadService } from '../../common/services/bulkhead.service';
     EventEmitterModule,
     QuotaModule,
     JobsModule,
+    StellarModule,
   ],
   controllers: [PayoutsController],
   providers: [