Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion src/clips/clips.module.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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: [
Expand All @@ -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: [
Expand All @@ -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 {}
98 changes: 98 additions & 0 deletions src/clips/nft-mint-enqueue.service.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
import { ConflictException, NotFoundException } from '@nestjs/common';
import { NftMintEnqueueService } from './nft-mint-enqueue.service';

function makeQueue() {
const jobs = new Map<string, any>();
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,
);
});
});
121 changes: 121 additions & 0 deletions src/clips/nft-mint-enqueue.service.ts
Original file line number Diff line number Diff line change
@@ -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<NftMintJob>,
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<any>,
jobName: 'mint',
data: {
clipId: input.clipId,
walletAddress: input.walletAddress,
userId: input.userId,
},
baseOptions: {
...NFT_MINT_JOB_OPTIONS,
jobId,
} as Record<string, unknown>,
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,
};
}
}
85 changes: 85 additions & 0 deletions src/nft/nft.controller.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -195,6 +196,7 @@ export class NftController {
private readonly nftMetadataRefreshService: NftMetadataRefreshService,
private readonly claimRoyaltyService: ClaimRoyaltyService,
private readonly royaltyClaimHistoryService: RoyaltyClaimHistoryService,
private readonly nftMintEnqueueService: NftMintEnqueueService,
) {}

@Auth()
Expand Down Expand Up @@ -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)
Expand Down
Loading