From 1cfc0b455267a9672b4e5e4d06bece63265ba042 Mon Sep 17 00:00:00 2001 From: GitHub Copilot Date: Fri, 2 Oct 2026 12:42:47 +0100 Subject: [PATCH] feat(nft): dedicated async mint queue with dedupe and status tracking (#974) --- src/clips/clips.module.ts | 10 +- src/clips/nft-mint-enqueue.service.spec.ts | 98 +++++++++++++++++ src/clips/nft-mint-enqueue.service.ts | 121 +++++++++++++++++++++ src/nft/nft.controller.ts | 85 +++++++++++++++ 4 files changed, 313 insertions(+), 1 deletion(-) create mode 100644 src/clips/nft-mint-enqueue.service.spec.ts create mode 100644 src/clips/nft-mint-enqueue.service.ts diff --git a/src/clips/clips.module.ts b/src/clips/clips.module.ts index d995c1a..779be90 100644 --- a/src/clips/clips.module.ts +++ b/src/clips/clips.module.ts @@ -13,6 +13,10 @@ import { NftMetadataService } from '../nft/nft-metadata.service'; import { RoyaltyConfigurationService } from '../nft/royalty-configuration.service'; import { registerQueue } from '../common'; import { CLIP_GENERATION_QUEUE } from './clip-generation.queue'; +import { NFT_MINT_QUEUE } from './nft-mint.queue'; +import { NftMintProcessor } from './nft-mint.processor'; +import { NftMintEnqueueService } from './nft-mint-enqueue.service'; +import { QueueOverflowService } from '../common/queue/queue-overflow.service'; @Module({ imports: [ @@ -21,6 +25,7 @@ import { CLIP_GENERATION_QUEUE } from './clip-generation.queue'; CircuitBreakerModule, IpfsUploadModule, registerQueue(CLIP_GENERATION_QUEUE), + registerQueue(NFT_MINT_QUEUE), ], controllers: [ClipsController], providers: [ @@ -30,8 +35,11 @@ import { CLIP_GENERATION_QUEUE } from './clip-generation.queue'; NftConfig, NftMetadataService, NftMintService, + NftMintEnqueueService, + NftMintProcessor, RoyaltyConfigurationService, + QueueOverflowService, ], - exports: [ClipsService, CloudinaryService, NftMintService], + exports: [ClipsService, CloudinaryService, NftMintService, NftMintEnqueueService], }) export class ClipsModule {} diff --git a/src/clips/nft-mint-enqueue.service.spec.ts b/src/clips/nft-mint-enqueue.service.spec.ts new file mode 100644 index 0000000..def89da --- /dev/null +++ b/src/clips/nft-mint-enqueue.service.spec.ts @@ -0,0 +1,98 @@ +import { ConflictException, NotFoundException } from '@nestjs/common'; +import { NftMintEnqueueService } from './nft-mint-enqueue.service'; + +function makeQueue() { + const jobs = new Map(); + return { + jobs, + getJob: jest.fn(async (id: string) => jobs.get(id)), + add: jest.fn(async (_name: string, _data: any, opts: any) => { + const job = { + id: opts.jobId, + attemptsMade: 0, + getState: async () => 'waiting', + }; + jobs.set(opts.jobId, job); + return job; + }), + }; +} + +function makeService() { + const queue = makeQueue(); + const overflowService = { + enqueue: jest.fn(async (o: any) => + queue.add(o.jobName, o.data, o.baseOptions), + ), + }; + const configService = { get: (_k: string, fb: string) => fb }; + const service = new NftMintEnqueueService( + queue as any, + overflowService as any, + configService as any, + ); + return { service, queue, overflowService }; +} + +describe('NftMintEnqueueService (#974)', () => { + it('uses a deterministic jobId per clip', async () => { + const { service, queue } = makeService(); + const res = await service.enqueueMint({ + clipId: 42, + walletAddress: 'GABC', + userId: 7, + }); + expect(res.jobId).toBe('nft-mint-clip-42'); + expect(queue.add).toHaveBeenCalledWith( + 'mint', + expect.objectContaining({ clipId: 42 }), + expect.objectContaining({ jobId: 'nft-mint-clip-42' }), + ); + }); + + it('prevents duplicate active mint jobs with 409', async () => { + const { service } = makeService(); + await service.enqueueMint({ clipId: 42, walletAddress: 'GABC', userId: 7 }); + await expect( + service.enqueueMint({ clipId: 42, walletAddress: 'GABC', userId: 7 }), + ).rejects.toBeInstanceOf(ConflictException); + }); + + it('allows re-enqueue after the previous job completed', async () => { + const { service, queue } = makeService(); + await service.enqueueMint({ clipId: 42, walletAddress: 'GABC', userId: 7 }); + const job = queue.jobs.get('nft-mint-clip-42'); + job.getState = async () => 'completed'; + const res = await service.enqueueMint({ + clipId: 42, + walletAddress: 'GABC', + userId: 7, + }); + expect(res.jobId).toBe('nft-mint-clip-42'); + }); + + it('maps job states to transaction states', async () => { + const { service, queue } = makeService(); + await service.enqueueMint({ clipId: 1, walletAddress: 'GABC', userId: 7 }); + const waiting = await service.getMintStatus('nft-mint-clip-1'); + expect(waiting.txState).toBe('pending'); + + const job = queue.jobs.get('nft-mint-clip-1'); + job.getState = async () => 'completed'; + job.returnvalue = { xdr: 'AAAA' }; + expect((await service.getMintStatus('nft-mint-clip-1')).txState).toBe('confirmed'); + + job.getState = async () => 'failed'; + job.failedReason = 'boom'; + const failed = await service.getMintStatus('nft-mint-clip-1'); + expect(failed.txState).toBe('failed'); + expect(failed.failedReason).toBe('boom'); + }); + + it('throws 404 for unknown jobId', async () => { + const { service } = makeService(); + await expect(service.getMintStatus('nft-mint-clip-999')).rejects.toBeInstanceOf( + NotFoundException, + ); + }); +}); diff --git a/src/clips/nft-mint-enqueue.service.ts b/src/clips/nft-mint-enqueue.service.ts new file mode 100644 index 0000000..09ce491 --- /dev/null +++ b/src/clips/nft-mint-enqueue.service.ts @@ -0,0 +1,121 @@ +import { + Injectable, + Logger, + ConflictException, + NotFoundException, +} from '@nestjs/common'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Queue } from 'bullmq'; +import { ConfigService } from '@nestjs/config'; +import { NFT_MINT_QUEUE, NFT_MINT_JOB_OPTIONS } from './nft-mint.queue'; +import type { NftMintJob } from './nft-mint.processor'; +import { QueueOverflowService } from '../common/queue/queue-overflow.service'; +import { getBullMQRateLimitConfig } from '../config/bullmq.config'; + +export type NftMintTxState = + | 'pending' + | 'confirmed' + | 'failed' + | 'unknown'; + +/** + * NftMintEnqueueService (#974) + * + * Moves Soroban mint processing off the request path and into the dedicated + * `nft-mint` BullMQ queue (separate worker, own concurrency + retry policy), + * so video processing never blocks NFT minting and vice versa. + * + * Dedupe: the BullMQ jobId is deterministic (`nft-mint-{clipId}`), so a + * second enqueue for a clip with an active job returns 409 instead of + * double-minting. + */ +@Injectable() +export class NftMintEnqueueService { + private readonly logger = new Logger(NftMintEnqueueService.name); + + constructor( + @InjectQueue(NFT_MINT_QUEUE) private readonly mintQueue: Queue, + private readonly queueOverflowService: QueueOverflowService, + private readonly configService: ConfigService, + ) {} + + static jobIdForClip(clipId: number): string { + return `nft-mint-clip-${clipId}`; + } + + async enqueueMint(input: { + clipId: number; + walletAddress: string; + userId: number; + }): Promise<{ jobId: string; delayed: boolean; delayMs: number }> { + const jobId = NftMintEnqueueService.jobIdForClip(input.clipId); + + // Dedupe: an already-queued/active job for this clip is a 409. + const existing = await this.mintQueue.getJob(jobId); + if (existing) { + const state = await existing.getState().catch(() => 'unknown'); + if (state === 'waiting' || state === 'active' || state === 'delayed' || state === 'prioritized') { + throw new ConflictException( + `Mint job for clip ${input.clipId} is already ${state} (job ${jobId})`, + ); + } + } + + const rateLimits = getBullMQRateLimitConfig(this.configService); + const result = await this.queueOverflowService.enqueue({ + queue: this.mintQueue as Queue, + jobName: 'mint', + data: { + clipId: input.clipId, + walletAddress: input.walletAddress, + userId: input.userId, + }, + baseOptions: { + ...NFT_MINT_JOB_OPTIONS, + jobId, + } as Record, + rateLimitConfig: rateLimits.nftMint, + }); + + this.logger.log( + `Enqueued NFT mint job ${result.jobId} for clip ${input.clipId}` + + (result.delayed ? ` (delayed ${result.delayMs}ms by overflow guard)` : ''), + ); + return { + jobId: result.jobId ?? jobId, + delayed: result.delayed, + delayMs: result.delayMs, + }; + } + + async getMintStatus(jobId: string): Promise<{ + jobId: string; + state: string; + txState: NftMintTxState; + attemptsMade: number; + failedReason?: string; + result?: unknown; + }> { + const job = await this.mintQueue.getJob(jobId); + if (!job) { + throw new NotFoundException(`Mint job ${jobId} not found`); + } + const state = await job.getState(); + const txState: NftMintTxState = + state === 'completed' + ? 'confirmed' + : state === 'failed' + ? 'failed' + : state === 'waiting' || state === 'active' || state === 'delayed' || state === 'prioritized' + ? 'pending' + : 'unknown'; + return { + jobId, + state, + txState, + attemptsMade: job.attemptsMade, + failedReason: job.failedReason, + result: job.returnvalue, + }; + } +} diff --git a/src/nft/nft.controller.ts b/src/nft/nft.controller.ts index e58e2d6..e6041e3 100644 --- a/src/nft/nft.controller.ts +++ b/src/nft/nft.controller.ts @@ -139,6 +139,7 @@ import { CollectionInfoResponseDto } from './dto/collection-info.dto'; import { TotalSupplyResponseDto } from './dto/total-supply.dto'; import { GasStatsResponseDto } from './dto/gas-stats.dto'; import { GasMetricsService } from './gas-metrics.service'; +import { NftMintEnqueueService } from '../clips/nft-mint-enqueue.service'; import { UpdateMetadataDto, UpdateMetadataResponseDto, @@ -195,6 +196,7 @@ export class NftController { private readonly nftMetadataRefreshService: NftMetadataRefreshService, private readonly claimRoyaltyService: ClaimRoyaltyService, private readonly royaltyClaimHistoryService: RoyaltyClaimHistoryService, + private readonly nftMintEnqueueService: NftMintEnqueueService, ) {} @Auth() @@ -758,6 +760,89 @@ export class NftController { ); } + @Auth() + @UseGuards(NftMintGuard, QueueRateLimitGuard) + @Post('enqueue-mint') + @HttpCode(HttpStatus.ACCEPTED) + @Throttle({ nftMint: { limit: 5, ttl: 60000 } }) + @QueueRateLimit({ queue: 'nft-mint', maxJobs: 5 }) + @ApiBearerAuth('access-token') + @ApiOperation({ + summary: 'Enqueue an NFT mint job on the dedicated nft-mint queue', + description: + 'Moves Soroban mint processing off the request path into the dedicated nft-mint BullMQ queue ' + + '(own worker, concurrency and retry policy), so video processing never blocks minting. ' + + 'Returns the BullMQ jobId immediately; use GET /nfts/mint-status/:jobId to track transaction states. ' + + 'Duplicate active jobs for the same clip return 409.', + }) + @ApiBody({ type: MintNftDto }) + @ApiResponse({ + status: 202, + description: 'Mint job enqueued; returns jobId for status tracking', + schema: { + example: { jobId: 'nft-mint-clip-42', delayed: false, delayMs: 0 }, + }, + }) + @ApiBadRequestResponse({ description: 'Invalid mint payload' }) + @ApiUnauthorizedResponse({ + description: 'Unauthorized — Bearer JWT required', + }) + @ApiForbiddenResponse({ + description: 'Caller does not own the clip', + }) + @ApiNotFoundResponse({ description: 'Clip not found' }) + @ApiConflictResponse({ + description: 'A mint job for this clip is already queued or running (duplicate prevented)', + type: NftMintConflictDto, + }) + @ApiTooManyRequestsResponse({ + description: + 'Too many active NFT mint jobs for this user. Retry after the number of seconds in the Retry-After header.', + }) + async enqueueMint( + @Body() dto: MintNftDto, + @Req() req: Request, + ): Promise<{ jobId: string; delayed: boolean; delayMs: number }> { + const userId = Number((req as any).user?.id ?? 0); + await this.nftMintService.validateClipOwner(dto.clipId, userId); + return this.nftMintEnqueueService.enqueueMint({ + clipId: dto.clipId, + walletAddress: dto.creatorWallet, + userId, + }); + } + + @Auth() + @Get('mint-status/:jobId') + @ApiBearerAuth('access-token') + @ApiOperation({ + summary: 'Get NFT mint job transaction status', + description: + 'Tracks a BullMQ mint job (returned by enqueue-mint) through pending → confirmed/failed states. ' + + 'pending covers waiting/active/delayed; confirmed means the mint transaction completed; ' + + 'failed includes the failure reason.', + }) + @ApiParam({ name: 'jobId', description: 'BullMQ job ID returned by enqueue-mint', example: 'nft-mint-clip-42' }) + @ApiResponse({ + status: 200, + description: 'Mint job status with transaction state', + schema: { + example: { + jobId: 'nft-mint-clip-42', + state: 'completed', + txState: 'confirmed', + attemptsMade: 1, + }, + }, + }) + @ApiUnauthorizedResponse({ + description: 'Unauthorized — Bearer JWT required', + }) + @ApiNotFoundResponse({ description: 'Mint job not found' }) + async mintJobStatus(@Param('jobId') jobId: string) { + return this.nftMintEnqueueService.getMintStatus(jobId); + } + @Auth() @Post('confirm-mint') @HttpCode(HttpStatus.OK)