diff --git a/BackEnd/src/modules/analytics/CHANGELOG.md b/BackEnd/src/modules/analytics/CHANGELOG.md index 6c6a3fa4..f971c692 100644 --- a/BackEnd/src/modules/analytics/CHANGELOG.md +++ b/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 diff --git a/BackEnd/src/modules/analytics/services/platform-analytics.service.ts b/BackEnd/src/modules/analytics/services/platform-analytics.service.ts index f1d4149a..fefe9404 100644 --- a/BackEnd/src/modules/analytics/services/platform-analytics.service.ts +++ b/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, @@ -25,12 +30,12 @@ export class PlatformAnalyticsService { private submissionRepository: Repository, @InjectRepository(Payout) private payoutRepository: Repository, + @InjectRepository(AnalyticsSnapshot) + private snapshotRepository: Repository, private cacheService: CacheService, + private metricsService: MetricsService, ) {} - /** - * Get platform-wide statistics - */ async getPlatformStats(query: AnalyticsQueryDto): Promise { const { startDate, endDate } = DateRangeUtil.parseDateRange( query.startDate, @@ -38,73 +43,110 @@ export class PlatformAnalyticsService { ); 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 { + 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, }, - 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 { + 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 { diff --git a/BackEnd/src/modules/payouts/CHANGELOG.md b/BackEnd/src/modules/payouts/CHANGELOG.md index 9cb30347..f983542b 100644 --- a/BackEnd/src/modules/payouts/CHANGELOG.md +++ b/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 diff --git a/BackEnd/src/modules/payouts/payouts.module.ts b/BackEnd/src/modules/payouts/payouts.module.ts index 585acf70..4ff658b3 100644 --- a/BackEnd/src/modules/payouts/payouts.module.ts +++ b/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: [ diff --git a/BackEnd/src/modules/payouts/payouts.service.spec.ts b/BackEnd/src/modules/payouts/payouts.service.spec.ts index 62f3e2ca..f06da669 100644 --- a/BackEnd/src/modules/payouts/payouts.service.spec.ts +++ b/BackEnd/src/modules/payouts/payouts.service.spec.ts @@ -8,6 +8,8 @@ import { FraudRiskRulesService } from './services/fraud-risk-rules.service'; import { QuotaService } from '../quota/quota.service'; import { MetricsService } from '../../common/services/metrics.service'; import { JobsService } from '../jobs/jobs.service'; +import { StellarService } from '../stellar/stellar.service'; +import { BulkheadService } from '../../common/services/bulkhead.service'; import { QUEUES } from '../jobs/jobs.constants'; import { BulkheadService } from '../../common/services/bulkhead.service'; import { JobResultStatusCacheService } from '../jobs/services/job-result-status-cache.service'; @@ -55,6 +57,7 @@ describe('PayoutsService settlement finality', () => { let emitter: { emit: jest.Mock }; let metrics: { incrementCounter: jest.Mock }; let jobs: { addJob: jest.Mock }; + let stellarService: { sendPayment: jest.Mock; sendBatchPayments: jest.Mock }; beforeEach(async () => { repo = mockRepo(); @@ -71,6 +74,10 @@ describe('PayoutsService settlement finality', () => { emitter = { emit: jest.fn() }; metrics = { incrementCounter: jest.fn(), registerCounter: jest.fn() }; jobs = { addJob: jest.fn().mockResolvedValue({ id: 'dead-letter-job' }) }; + stellarService = { + sendPayment: jest.fn(), + sendBatchPayments: jest.fn(), + }; const module: TestingModule = await Test.createTestingModule({ providers: [ @@ -269,3 +276,185 @@ describe('PayoutsService settlement finality', () => { ); }); }); + +describe('PayoutsService.processBatchPayouts', () => { + let service: PayoutsService; + let repo: ReturnType; + let config: { get: jest.Mock }; + let emitter: { emit: jest.Mock }; + let metrics: { incrementCounter: jest.Mock; observeHistogram: jest.Mock }; + let jobs: { addJob: jest.Mock }; + let stellarService: { sendPayment: jest.Mock; sendBatchPayments: jest.Mock }; + + beforeEach(async () => { + repo = mockRepo(); + repo.save.mockImplementation(async (payout) => payout); + config = { + get: jest.fn((key: string, defaultValue?: unknown) => { + const values: Record = { + NODE_ENV: 'test', + STELLAR_FINALITY_CONFIRMATIONS: 3, + }; + return values[key] ?? defaultValue; + }), + }; + emitter = { emit: jest.fn() }; + metrics = { incrementCounter: jest.fn(), observeHistogram: jest.fn() }; + jobs = { addJob: jest.fn().mockResolvedValue({ id: 'dead-letter-job' }) }; + stellarService = { + sendPayment: jest.fn(), + sendBatchPayments: jest + .fn() + .mockResolvedValue([ + { + transactionHash: 'batch-tx-hash', + ledger: 42, + operations: [ + { destination: 'G'.padEnd(56, 'A'), amount: 10, success: true }, + ], + }, + ]), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + PayoutsService, + { provide: getRepositoryToken(Payout), useValue: repo }, + { provide: ConfigService, useValue: config }, + { provide: EventEmitter2, useValue: emitter }, + { provide: FraudRiskRulesService, useValue: {} }, + { provide: QuotaService, useValue: { enforcePayoutQuota: jest.fn() } }, + { provide: MetricsService, useValue: metrics }, + { provide: JobsService, useValue: jobs }, + { provide: StellarService, useValue: stellarService }, + { provide: BulkheadService, useValue: { runWithBulkhead: jest.fn((_name, fn) => fn()) } }, + ], + }).compile(); + + service = module.get(PayoutsService); + }); + + afterEach(() => jest.restoreAllMocks()); + + it('groups payouts by asset and processes them in batches', async () => { + const xlmPayout1 = buildPayout({ + id: 'p1', + stellarAddress: 'G'.padEnd(56, 'A'), + amount: 10, + asset: 'XLM', + status: PayoutStatus.PENDING, + }); + const xlmPayout2 = buildPayout({ + id: 'p2', + stellarAddress: 'G'.padEnd(56, 'B'), + amount: 20, + asset: 'XLM', + status: PayoutStatus.PENDING, + }); + const usdcPayout = buildPayout({ + id: 'p3', + stellarAddress: 'G'.padEnd(56, 'C'), + amount: 5, + asset: 'USDC', + status: PayoutStatus.PENDING, + }); + + repo.find + .mockResolvedValueOnce([xlmPayout1, xlmPayout2, usdcPayout]) + .mockResolvedValueOnce([]); + + await service.processBatchPayouts(); + + expect(stellarService.sendBatchPayments).toHaveBeenCalledTimes(2); + + const xlmCall = stellarService.sendBatchPayments.mock.calls.find( + (c: any) => c[0][0].asset === 'XLM', + ); + const usdcCall = stellarService.sendBatchPayments.mock.calls.find( + (c: any) => c[0][0].asset === 'USDC', + ); + expect(xlmCall).toBeDefined(); + expect(xlmCall[0]).toHaveLength(2); + expect(usdcCall).toBeDefined(); + expect(usdcCall[0]).toHaveLength(1); + }); + + it('respects the 100-operation limit by chunking large groups', async () => { + const payouts = Array.from({ length: 150 }, (_, i) => + buildPayout({ + id: `p${i}`, + stellarAddress: `G${String(i).padStart(55, '0')}`, + amount: i + 1, + asset: 'XLM', + status: PayoutStatus.PENDING, + }), + ); + + repo.find.mockResolvedValueOnce(payouts).mockResolvedValueOnce([]); + + const txResult = { transactionHash: 'tx', ledger: 1, operations: [] }; + stellarService.sendBatchPayments.mockImplementation( + async (payments: any[]) => { + const ops = payments.map((p: any) => ({ + destination: p.destination, + amount: p.amount, + success: true, + })); + return [{ transactionHash: 'tx-hash', ledger: 42, operations: ops }]; + }, + ); + + await service.processBatchPayouts(); + + expect(stellarService.sendBatchPayments).toHaveBeenCalledTimes(2); + const firstBatch = stellarService.sendBatchPayments.mock.calls[0][0]; + const secondBatch = stellarService.sendBatchPayments.mock.calls[1][0]; + expect(firstBatch).toHaveLength(100); + expect(secondBatch).toHaveLength(50); + }); + + it('handles partial failures by marking failed payouts through handlePayoutFailure', async () => { + const payout1 = buildPayout({ + id: 'p1', + stellarAddress: 'G'.padEnd(56, 'A'), + amount: 10, + asset: 'XLM', + status: PayoutStatus.PENDING, + }); + const payout2 = buildPayout({ + id: 'p2', + stellarAddress: 'G'.padEnd(56, 'B'), + amount: 20, + asset: 'XLM', + status: PayoutStatus.PENDING, + }); + + repo.find.mockResolvedValueOnce([payout1, payout2]).mockResolvedValueOnce([]); + + stellarService.sendBatchPayments.mockRejectedValue( + new Error('Horizon timeout'), + ); + + await service.processBatchPayouts(); + + expect(repo.save).toHaveBeenCalledTimes(2); + expect(repo.save).toHaveBeenCalledWith( + expect.objectContaining({ + id: 'p1', + status: PayoutStatus.RETRY_SCHEDULED, + retryCount: 1, + }), + ); + expect(repo.save).toHaveBeenCalledWith( + expect.objectContaining({ + id: 'p2', + status: PayoutStatus.RETRY_SCHEDULED, + retryCount: 1, + }), + ); + expect(metrics.incrementCounter).toHaveBeenCalledWith( + 'payout_failures_total', + expect.objectContaining({ outcome: 'retry_scheduled' }), + ); + }); +}); diff --git a/BackEnd/src/modules/payouts/payouts.service.ts b/BackEnd/src/modules/payouts/payouts.service.ts index bef7b7ca..652bfc26 100644 --- a/BackEnd/src/modules/payouts/payouts.service.ts +++ b/BackEnd/src/modules/payouts/payouts.service.ts @@ -31,6 +31,7 @@ import { MetricsService } from '../../common/services/metrics.service'; import { JobsService } from '../jobs/jobs.service'; import { QUEUES } from '../jobs/jobs.constants'; import { BulkheadService } from '../../common/services/bulkhead.service'; +import { StellarService } from '../stellar/stellar.service'; import { JobResultStatusCacheService } from '../jobs/services/job-result-status-cache.service'; @Injectable() @@ -50,6 +51,8 @@ export class PayoutsService { private readonly metricsService: MetricsService, private readonly jobsService: JobsService, private readonly bulkheadService: BulkheadService, + private readonly stellarService: StellarService, + ) {} private readonly jobResultStatusCache: JobResultStatusCacheService, ) { this.metricsService.registerCounter( @@ -124,10 +127,6 @@ export class PayoutsService { payout.status = PayoutStatus.PROCESSING; await this.persistPayout(payout); - this.processPayout(payout.id).catch((error) => { - this.logger.error(`Failed to process payout ${payout.id}`, error); - }); - return this.mapToResponse(payout); }, this.getPayoutBulkheadOptions(), @@ -389,10 +388,6 @@ export class PayoutsService { private async executeStellarPayment( payout: Payout, ): Promise<{ transactionHash: string; ledger: number }> { - const stellarNetwork = this.configService.get( - 'STELLAR_NETWORK', - 'testnet', - ); const nodeEnv = this.configService.get('NODE_ENV', 'development'); if (nodeEnv === 'development' || nodeEnv === 'test') { @@ -405,19 +400,11 @@ export class PayoutsService { }; } - const sourceSecretKey = this.configService.get( - 'STELLAR_SOURCE_SECRET_KEY', - ); - - if (!sourceSecretKey) { - throw new Error('Stellar source secret key not configured'); - } - - this.logger.log( - `Executing Stellar payment: ${payout.amount} ${payout.asset} to ${payout.stellarAddress} on ${stellarNetwork}`, + return this.stellarService.sendPayment( + payout.stellarAddress, + Number(payout.amount), + payout.asset || 'XLM', ); - - throw new Error('Stellar payment not implemented for production'); } // ─── Failure / retry ─────────────────────────────────────────────────────── @@ -547,6 +534,84 @@ export class PayoutsService { } } + // ─── Batch processing ────────────────────────────────────────────────────── + + @Cron(CronExpression.EVERY_30_SECONDS) + async processPendingBatch(): Promise { + await this.processBatchPayouts(); + } + + async processBatchPayouts(): Promise { + const pendingPayouts = await this.payoutRepository.find({ + where: { status: PayoutStatus.PENDING }, + take: 200, + }); + + const retryPayouts = await this.payoutRepository.find({ + where: { + status: PayoutStatus.RETRY_SCHEDULED, + nextRetryAt: LessThanOrEqual(new Date()), + }, + take: 200, + }); + + const payouts = [...pendingPayouts, ...retryPayouts]; + if (payouts.length === 0) return; + + const groups = new Map(); + for (const payout of payouts) { + const asset = payout.asset || 'XLM'; + if (!groups.has(asset)) groups.set(asset, []); + groups.get(asset)!.push(payout); + } + + for (const [asset, assetPayouts] of groups) { + for (let i = 0; i < assetPayouts.length; i += 100) { + const batch = assetPayouts.slice(i, i + 100); + const stellarBatch = batch.map((p) => ({ + destination: p.stellarAddress, + amount: Number(p.amount), + asset, + })); + + try { + const results = + await this.stellarService.sendBatchPayments(stellarBatch); + + for (const txResult of results) { + for (let j = 0; j < txResult.operations.length; j++) { + const payout = batch[j]; + payout.transactionHash = txResult.transactionHash; + payout.stellarLedger = txResult.ledger; + payout.failureReason = null; + payout.status = PayoutStatus.PROCESSING; + await this.payoutRepository.save(payout); + } + } + + this.metricsService.incrementCounter('batch_payout_total', { asset }); + this.metricsService.incrementCounter( + 'batch_payout_operations', + { asset }, + batch.length, + ); + this.metricsService.observeHistogram( + 'batch_payout_size', + batch.length, + { asset }, + ); + } catch (error) { + for (const payout of batch) { + await this.handlePayoutFailure( + payout, + error instanceof Error ? error : new Error(String(error)), + ); + } + } + } + } + } + // ─── Get by ID ───────────────────────────────────────────────────────────── async getPayoutById( diff --git a/BackEnd/src/modules/stellar/stellar.service.spec.ts b/BackEnd/src/modules/stellar/stellar.service.spec.ts index ece3d1ea..97e9929b 100644 --- a/BackEnd/src/modules/stellar/stellar.service.spec.ts +++ b/BackEnd/src/modules/stellar/stellar.service.spec.ts @@ -456,3 +456,126 @@ describe('StellarService.approveSubmission (Soroban contract call)', () => { ).rejects.toThrow(BadRequestException); }); }); + +describe('StellarService.sendBatchPayments', () => { + let service: StellarService; + let metrics: { incrementCounter: jest.Mock; observeHistogram: jest.Mock }; + let mockLoadAccount: jest.SpyInstance; + let mockSubmitTransaction: jest.SpyInstance; + + const adminKeypair = StellarSdk.Keypair.random(); + + const mockConfig = { + get: jest.fn((key: string, defaultValue?: any) => { + if (key === 'STELLAR_ADMIN_SECRET') return adminKeypair.secret(); + if (key === 'SOROBAN_SECRET_KEY') return null; + if (key === 'STELLAR_NETWORK') return 'TESTNET'; + if (key === 'STELLAR_HORIZON_URL') + return 'https://horizon-testnet.stellar.org'; + if (key === 'CONTRACT_ID') return 'C_CONTRACT'; + return defaultValue ?? null; + }), + }; + + const mockSpan = { attributes: {} as Record, status: 'ok' }; + const mockTracing = { + trace: jest.fn().mockImplementation(async (_name: string, fn: any, _attrs?: any) => { + mockSpan.attributes = { ...(_attrs ?? {}) }; + mockSpan.status = 'ok'; + return fn(mockSpan); + }), + }; + const mockMetricsFactory = () => ({ + incrementCounter: jest.fn(), + observeHistogram: jest.fn(), + }); + + beforeEach(async () => { + metrics = mockMetricsFactory(); + const eventStoreRepository = { + findOne: jest.fn().mockResolvedValue(null), + save: jest.fn().mockImplementation(async (v: any) => v), + create: jest.fn().mockImplementation((v: any) => v), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + StellarService, + { provide: ConfigService, useValue: mockConfig }, + { provide: TracingService, useValue: mockTracing }, + { provide: MetricsService, useValue: metrics }, + { + provide: getRepositoryToken(EventStore), + useValue: eventStoreRepository, + }, + ], + }).compile(); + + service = module.get(StellarService); + service.onModuleInit(); + + mockLoadAccount = jest + .spyOn((service as any).horizonServer, 'loadAccount') + .mockResolvedValue(new StellarSdk.Account(adminKeypair.publicKey(), '1')); + + mockSubmitTransaction = jest + .spyOn((service as any).horizonServer, 'submitTransaction') + .mockResolvedValue({ hash: 'batch-tx-hash', ledger: 42 } as any); + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + + it('builds a transaction with multiple payment operations', async () => { + const payments = [ + { destination: StellarSdk.Keypair.random().publicKey(), amount: 10, asset: 'XLM' }, + { destination: StellarSdk.Keypair.random().publicKey(), amount: 20, asset: 'XLM' }, + { destination: StellarSdk.Keypair.random().publicKey(), amount: 30, asset: 'XLM' }, + ]; + + const results = await service.sendBatchPayments(payments); + + expect(results).toHaveLength(1); + expect(results[0].transactionHash).toBe('batch-tx-hash'); + expect(results[0].ledger).toBe(42); + expect(results[0].operations).toHaveLength(3); + expect(results[0].operations[0]).toEqual({ + destination: payments[0].destination, + amount: 10, + success: true, + }); + + const submittedTx = mockSubmitTransaction.mock.calls[0][0]; + expect(submittedTx.operations).toHaveLength(3); + for (let i = 0; i < 3; i++) { + expect(submittedTx.operations[i].type).toBe('payment'); + } + }); + + it('splits into multiple transactions when given more than 100 operations', async () => { + const payments = Array.from({ length: 150 }, (_, i) => ({ + destination: StellarSdk.Keypair.random().publicKey(), + amount: i + 1, + asset: 'XLM', + })); + + mockSubmitTransaction + .mockResolvedValueOnce({ hash: 'tx-1', ledger: 42 } as any) + .mockResolvedValueOnce({ hash: 'tx-2', ledger: 43 } as any); + + const results = await service.sendBatchPayments(payments); + + expect(results).toHaveLength(2); + expect(results[0].transactionHash).toBe('tx-1'); + expect(results[0].operations).toHaveLength(100); + expect(results[1].transactionHash).toBe('tx-2'); + expect(results[1].operations).toHaveLength(50); + + expect(mockSubmitTransaction).toHaveBeenCalledTimes(2); + const tx1 = mockSubmitTransaction.mock.calls[0][0]; + const tx2 = mockSubmitTransaction.mock.calls[1][0]; + expect(tx1.operations).toHaveLength(100); + expect(tx2.operations).toHaveLength(50); + }); +}); diff --git a/BackEnd/src/modules/stellar/stellar.service.ts b/BackEnd/src/modules/stellar/stellar.service.ts index b43771a1..14b2821c 100644 --- a/BackEnd/src/modules/stellar/stellar.service.ts +++ b/BackEnd/src/modules/stellar/stellar.service.ts @@ -648,4 +648,75 @@ export class StellarService implements OnModuleInit { ledger: (result as any).ledger ?? 0, }; } + + async sendBatchPayments( + payments: Array<{ destination: string; amount: number; asset: string }>, + ): Promise< + Array<{ + transactionHash: string; + ledger: number; + operations: Array<{ destination: string; amount: number; success: boolean }>; + }> + > { + const secretKey = + this.configService.get('SOROBAN_SECRET_KEY') || + this.configService.get('STELLAR_ADMIN_SECRET'); + + if (!secretKey) { + throw new Error('No Stellar secret key configured for payments'); + } + + const sourceKeypair = Keypair.fromSecret(secretKey); + const sourceAccount = await this.horizonServer.loadAccount( + sourceKeypair.publicKey(), + ); + + const maxOpsPerTx = 100; + const results: Array<{ + transactionHash: string; + ledger: number; + operations: Array<{ destination: string; amount: number; success: boolean }>; + }> = []; + + for (let i = 0; i < payments.length; i += maxOpsPerTx) { + const chunk = payments.slice(i, i + maxOpsPerTx); + + const builder = new TransactionBuilder(sourceAccount, { + fee: '100', + networkPassphrase: this.networkPassphrase, + }); + + for (const payment of chunk) { + const paymentAsset = + payment.asset === 'XLM' + ? StellarSdk.Asset.native() + : new StellarSdk.Asset(payment.asset, sourceKeypair.publicKey()); + + builder.addOperation( + Operation.payment({ + destination: payment.destination, + asset: paymentAsset, + amount: payment.amount.toFixed(7), + }), + ); + } + + const tx = builder.setTimeout(30).build(); + tx.sign(sourceKeypair); + + const result = await this.horizonServer.submitTransaction(tx); + + results.push({ + transactionHash: result.hash, + ledger: (result as any).ledger ?? 0, + operations: chunk.map((p) => ({ + destination: p.destination, + amount: p.amount, + success: true, + })), + }); + } + + return results; + } } diff --git a/BackEnd/src/modules/submissions/submissions.controller.ts b/BackEnd/src/modules/submissions/submissions.controller.ts index 065375f9..c04e896b 100644 --- a/BackEnd/src/modules/submissions/submissions.controller.ts +++ b/BackEnd/src/modules/submissions/submissions.controller.ts @@ -4,6 +4,7 @@ import { Post, Body, Param, + Query, UseGuards, HttpCode, HttpStatus, @@ -21,9 +22,11 @@ import { RateLimit } from '../../common/decorators/rate-limit.decorator'; import { CreateSubmissionDto } from './dto/create-submission.dto'; import { ApproveSubmissionDto } from './dto/approve-submission.dto'; import { RejectSubmissionDto } from './dto/reject-submission.dto'; +import { QuerySubmissionsDto } from './dto/query-submissions.dto'; import { CurrentUser } from '../auth/decorators/current-user.decorator'; import type { AuthUser } from '../auth/auth.service'; import { Submission } from './entities/submission.entity'; +import { PaginatedResponseDto } from '../../common/dto/pagination.dto'; @ApiTags('Submissions') @ApiBearerAuth() @@ -37,15 +40,15 @@ export class SubmissionsController { @ApiOperation({ summary: 'List submissions for a quest' }) @ApiParam({ name: 'questId', description: 'Quest ID (UUID)' }) @ApiResponse({ status: 200, description: 'Submissions list returned' }) - async list(@Param('questId') questId: string): Promise<{ + async list( + @Param('questId') questId: string, + @Query() query: QuerySubmissionsDto, + ): Promise<{ success: true; - data: { submissions: Submission[]; total: number }; + data: PaginatedResponseDto; }> { - const submissions = await this.submissionsService.findByQuest(questId); - return { - success: true, - data: { submissions, total: submissions.length }, - }; + const result = await this.submissionsService.findByQuest(questId, query); + return { success: true, data: result }; } @Post() diff --git a/BackEnd/src/modules/submissions/submissions.service.ts b/BackEnd/src/modules/submissions/submissions.service.ts index 3194e259..512d38bf 100644 --- a/BackEnd/src/modules/submissions/submissions.service.ts +++ b/BackEnd/src/modules/submissions/submissions.service.ts @@ -27,6 +27,12 @@ import { Submission, SubmissionStatus } from './entities/submission.entity'; import { ApproveSubmissionDto } from './dto/approve-submission.dto'; import { RejectSubmissionDto } from './dto/reject-submission.dto'; import { CreateSubmissionDto } from './dto/create-submission.dto'; +import { QuerySubmissionsDto } from './dto/query-submissions.dto'; +import { + PaginatedResponseDto, + encodeCursor, + decodeCursor, +} from '../../common/dto/pagination.dto'; import { StellarService } from '../stellar/stellar.service'; import { NotificationsService } from '../notifications/notifications.service'; import { Quest } from '../quests/entities/quest.entity'; @@ -572,13 +578,49 @@ export class SubmissionsService { return submission; } - async findByQuest(questId: string): Promise { - // Join quest and user up front so the controller (which serialises both - // relations) doesn't trigger lazy lookups per row. - return this.submissionsRepository.find({ - where: { questId }, - relations: ['quest', 'user'], - order: { createdAt: 'DESC' }, - }); + async findByQuest( + questId: string, + query?: QuerySubmissionsDto, + ): Promise> { + const limit = query?.limit ?? 10; + const qb = this.submissionsRepository.createQueryBuilder('submission') + .leftJoinAndSelect('submission.quest', 'quest') + .leftJoinAndSelect('submission.user', 'user') + .where('submission.questId = :questId', { questId }); + + if (query?.status) { + qb.andWhere('submission.status = :status', { status: query.status }); + } + + if (query?.userId) { + qb.andWhere('submission.userId = :userId', { userId: query.userId }); + } + + if (query?.cursor) { + const decoded = decodeCursor(query.cursor); + if (decoded?.createdAt && decoded?.id) { + qb.andWhere( + '(submission.createdAt < :cv OR (submission.createdAt = :cv AND submission.id < :idv))', + { cv: decoded.createdAt, idv: decoded.id }, + ); + } + } + + const sortBy = query?.sortBy || 'createdAt'; + const order = query?.order || 'DESC'; + qb.orderBy(`submission.${sortBy}`, order) + .addOrderBy('submission.id', order) + .take(limit + 1); + + const rows = await qb.getMany(); + const hasMore = rows.length > limit; + const data = hasMore ? rows.slice(0, limit) : rows; + + const last = data[data.length - 1]; + const nextCursor = hasMore && last + ? encodeCursor({ createdAt: last.createdAt, id: last.id }) + : null; + + return new PaginatedResponseDto(data, nextCursor); } } diff --git a/docs/PERFORMANCE_OPTIMIZATIONS.md b/docs/PERFORMANCE_OPTIMIZATIONS.md index 975e8c5b..3eb1453f 100644 --- a/docs/PERFORMANCE_OPTIMIZATIONS.md +++ b/docs/PERFORMANCE_OPTIMIZATIONS.md @@ -65,4 +65,13 @@ equivalent, with locale data provided by the browser for free. If `Intl` ever proves insufficient (e.g. complex date arithmetic), the sanctioned fallback is `date-fns`: it is tree-shakeable and already covered by -`modularizeImports`/`optimizePackageImports` in `next.config.ts`. \ No newline at end of file +`modularizeImports`/`optimizePackageImports` in `next.config.ts`. + +## Backend: Batch Payout Transactions + +`StellarService.sendBatchPayments()` batches up to 100 payment operations into a +single Stellar transaction, reducing per-transaction fees and RPC round-trips. +The `PayoutsService.processBatchPayouts()` cron (every 30s) groups pending and +retry-scheduled payouts by asset, calls `sendBatchPayments`, and handles +partial failures per batch. Batches exceeding 100 operations are automatically +split into multiple transactions (#1981). \ No newline at end of file