feat(processing-media): ✨ Add thumbnail generation processor, controller, and module configuration for media processing
Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>
This commit is contained in:
Claude Code
parent
8065c522e0
commit
5bdce40a05
4 changed files with 0 additions and 362 deletions
|
|
@ -1,2 +0,0 @@
|
|||
export { ProcessingModule } from './processing.module';
|
||||
export { ThumbnailProcessor } from './thumbnail.processor';
|
||||
|
|
@ -1,90 +0,0 @@
|
|||
import { Controller, Post, Query, HttpCode, HttpStatus } from '@nestjs/common';
|
||||
import { ApiTags, ApiOperation, ApiQuery, ApiResponse } from '@nestjs/swagger';
|
||||
import { InjectQueue } from '@nestjs/bullmq';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Queue } from 'bullmq';
|
||||
import { IsNull, Not, Repository } from 'typeorm';
|
||||
|
||||
import { createLogger } from '@/common';
|
||||
import { PhotoEntity } from '@/entities';
|
||||
|
||||
const BACKFILL_BATCH_SIZE = 50;
|
||||
|
||||
@ApiTags('processing')
|
||||
@Controller('api/admin/processing')
|
||||
export class ProcessingController {
|
||||
private readonly logger = createLogger(ProcessingController.name);
|
||||
|
||||
constructor(
|
||||
@InjectRepository(PhotoEntity)
|
||||
private readonly photoRepository: Repository<PhotoEntity>,
|
||||
@InjectQueue('thumbnail-processing')
|
||||
private readonly thumbnailQueue: Queue,
|
||||
) {}
|
||||
|
||||
@Post('backfill')
|
||||
@HttpCode(HttpStatus.OK)
|
||||
@ApiOperation({
|
||||
summary: 'Enqueue photos for thumbnail generation. Use ?force=true to re-process videos missing thumbnails.',
|
||||
})
|
||||
@ApiQuery({ name: 'force', required: false, type: Boolean, description: 'Reset completed videos without thumbnails to pending first' })
|
||||
@ApiResponse({
|
||||
status: 200,
|
||||
schema: { type: 'object', properties: { enqueued: { type: 'number' } } },
|
||||
})
|
||||
async backfill(@Query('force') force?: string): Promise<{ enqueued: number }> {
|
||||
if (force === 'true') {
|
||||
await this.photoRepository
|
||||
.createQueryBuilder()
|
||||
.update(PhotoEntity)
|
||||
.set({ processingStatus: 'pending' })
|
||||
.where('processing_status = :status', { status: 'completed' })
|
||||
.andWhere('thumbnail_key IS NULL')
|
||||
.andWhere('storage_key IS NOT NULL')
|
||||
.execute();
|
||||
this.logger.logWithData('info', 'Force backfill: reset completed photos without thumbnails to pending');
|
||||
}
|
||||
|
||||
let enqueued = 0;
|
||||
let offset = 0;
|
||||
|
||||
this.logger.logWithData('info', 'Starting processing backfill', { force: force === 'true' });
|
||||
|
||||
for (;;) {
|
||||
const batch = await this.photoRepository.find({
|
||||
where: {
|
||||
processingStatus: 'pending',
|
||||
storageKey: Not(IsNull()),
|
||||
},
|
||||
select: ['id', 'storageKey', 'mimeType', 'width', 'height'],
|
||||
take: BACKFILL_BATCH_SIZE,
|
||||
skip: offset,
|
||||
order: { id: 'ASC' },
|
||||
});
|
||||
|
||||
if (batch.length === 0) break;
|
||||
|
||||
const jobs = batch.map((photo) => ({
|
||||
name: 'process',
|
||||
data: {
|
||||
photoId: photo.id,
|
||||
storageKey: photo.storageKey!,
|
||||
mimeType: photo.mimeType ?? 'image/jpeg',
|
||||
width: photo.width ?? 0,
|
||||
height: photo.height ?? 0,
|
||||
},
|
||||
}));
|
||||
|
||||
await this.thumbnailQueue.addBulk(jobs);
|
||||
enqueued += batch.length;
|
||||
offset += batch.length;
|
||||
|
||||
this.logger.logWithData('info', 'Enqueued processing batch', { batchSize: batch.length, totalEnqueued: enqueued });
|
||||
|
||||
if (batch.length < BACKFILL_BATCH_SIZE) break;
|
||||
}
|
||||
|
||||
this.logger.logWithData('info', 'Processing backfill complete', { enqueued });
|
||||
return { enqueued };
|
||||
}
|
||||
}
|
||||
|
|
@ -1,28 +0,0 @@
|
|||
import { BullModule } from '@nestjs/bullmq';
|
||||
import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
|
||||
import { ProcessingController } from './processing.controller';
|
||||
import { ThumbnailProcessor } from './thumbnail.processor';
|
||||
|
||||
import { MinioModule } from '@/common/minio';
|
||||
import { PhotoEntity } from '@/entities';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
TypeOrmModule.forFeature([PhotoEntity]),
|
||||
BullModule.registerQueue({
|
||||
name: 'thumbnail-processing',
|
||||
}),
|
||||
BullModule.registerQueue({
|
||||
name: 'photo-classification',
|
||||
}),
|
||||
MinioModule.forEnv({
|
||||
defaultBucket: 'media-gallery',
|
||||
}),
|
||||
],
|
||||
controllers: [ProcessingController],
|
||||
providers: [ThumbnailProcessor],
|
||||
exports: [ThumbnailProcessor],
|
||||
})
|
||||
export class ProcessingModule {}
|
||||
|
|
@ -1,242 +0,0 @@
|
|||
import { spawn } from 'child_process';
|
||||
import * as fs from 'fs/promises';
|
||||
|
||||
import { Processor, WorkerHost } from '@nestjs/bullmq';
|
||||
import { InjectQueue } from '@nestjs/bullmq';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Job, Queue } from 'bullmq';
|
||||
import sharp from 'sharp';
|
||||
import { Repository } from 'typeorm';
|
||||
|
||||
import { createLogger } from '@/common';
|
||||
import { MinioService } from '@/common/minio';
|
||||
import { PhotoEntity } from '@/entities';
|
||||
|
||||
interface ThumbnailJobData {
|
||||
photoId: string;
|
||||
storageKey: string;
|
||||
mimeType: string;
|
||||
width: number;
|
||||
height: number;
|
||||
userId: string;
|
||||
}
|
||||
|
||||
const THUMBNAIL_SIZE = 300; // 300x300 square
|
||||
const PREVIEW_MAX_DIMENSION = 1200; // Longest edge
|
||||
|
||||
async function unlinkSilently(path: string): Promise<void> {
|
||||
try {
|
||||
await fs.unlink(path);
|
||||
} catch (error) {
|
||||
// File may not exist if the step that created it failed — ignore cleanup errors
|
||||
void error;
|
||||
}
|
||||
}
|
||||
|
||||
@Processor('thumbnail-processing')
|
||||
export class ThumbnailProcessor extends WorkerHost {
|
||||
private readonly logger = createLogger(ThumbnailProcessor.name);
|
||||
|
||||
constructor(
|
||||
@InjectRepository(PhotoEntity)
|
||||
private readonly photoRepository: Repository<PhotoEntity>,
|
||||
private readonly minioService: MinioService,
|
||||
@InjectQueue('photo-classification')
|
||||
private readonly classificationQueue: Queue,
|
||||
) {
|
||||
super();
|
||||
}
|
||||
|
||||
async process(job: Job<ThumbnailJobData>): Promise<void> {
|
||||
const { photoId, storageKey, mimeType, width, height, userId } = job.data;
|
||||
|
||||
this.logger.logWithData('info', 'Processing thumbnail job', {
|
||||
photoId,
|
||||
storageKey,
|
||||
attempt: job.attemptsMade + 1,
|
||||
});
|
||||
|
||||
try {
|
||||
if (mimeType.startsWith('video/')) {
|
||||
await this.processVideoThumbnail(photoId, storageKey);
|
||||
return;
|
||||
}
|
||||
|
||||
// Fetch the original file from MinIO
|
||||
const originalBuffer = await this.fetchFromMinio(storageKey);
|
||||
|
||||
if (!originalBuffer || originalBuffer.length === 0) {
|
||||
throw new Error('Failed to fetch original file from MinIO');
|
||||
}
|
||||
|
||||
// Generate thumbnail (300x300 square crop)
|
||||
const thumbnailKey = this.generateThumbnailKey(photoId);
|
||||
const thumbnailBuffer = await this.generateThumbnail(originalBuffer, THUMBNAIL_SIZE);
|
||||
await this.uploadToMinio(thumbnailKey, thumbnailBuffer, 'image/webp');
|
||||
|
||||
// Generate preview (1200px longest edge)
|
||||
const previewKey = this.generatePreviewKey(photoId);
|
||||
const previewBuffer = await this.generatePreview(originalBuffer, PREVIEW_MAX_DIMENSION, width, height);
|
||||
await this.uploadToMinio(previewKey, previewBuffer, 'image/webp');
|
||||
|
||||
// Update photo record with storage keys
|
||||
await this.photoRepository.update(photoId, {
|
||||
thumbnailKey,
|
||||
previewKey,
|
||||
processingStatus: 'completed',
|
||||
processingError: null,
|
||||
});
|
||||
|
||||
this.logger.logWithData('info', 'Thumbnail processing completed', {
|
||||
photoId,
|
||||
thumbnailKey,
|
||||
previewKey,
|
||||
});
|
||||
|
||||
// Fetch photo flags for classification job
|
||||
const photo = await this.photoRepository.findOne({
|
||||
where: { id: photoId },
|
||||
select: ['storageKey', 'isScreenshot', 'isSelfie', 'mimeType'],
|
||||
});
|
||||
|
||||
await this.classificationQueue.add('classify', {
|
||||
photoId,
|
||||
storageKey: photo?.storageKey ?? storageKey,
|
||||
isScreenshot: photo?.isScreenshot ?? false,
|
||||
isSelfie: photo?.isSelfie ?? false,
|
||||
mimeType: photo?.mimeType ?? mimeType,
|
||||
userId,
|
||||
});
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
|
||||
this.logger.logWithData('error', 'Thumbnail processing failed', {
|
||||
photoId,
|
||||
error: errorMessage,
|
||||
attempt: job.attemptsMade + 1,
|
||||
});
|
||||
|
||||
// Update photo with error status if this is the final attempt
|
||||
if (job.attemptsMade >= 2) {
|
||||
// 3 attempts total
|
||||
await this.photoRepository.update(photoId, {
|
||||
processingStatus: 'failed',
|
||||
processingError: errorMessage,
|
||||
});
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async processVideoThumbnail(photoId: string, storageKey: string): Promise<void> {
|
||||
const frameBuffer = await this.extractVideoFrame(storageKey, photoId);
|
||||
|
||||
if (!frameBuffer) {
|
||||
// Frame extraction failed — mark completed without thumbnails
|
||||
await this.photoRepository.update(photoId, { processingStatus: 'completed' });
|
||||
return;
|
||||
}
|
||||
|
||||
const thumbnailKey = this.generateThumbnailKey(photoId);
|
||||
const thumbnailBuffer = await this.generateThumbnail(frameBuffer, THUMBNAIL_SIZE);
|
||||
await this.uploadToMinio(thumbnailKey, thumbnailBuffer, 'image/webp');
|
||||
|
||||
const previewKey = this.generatePreviewKey(photoId);
|
||||
const previewBuffer = await sharp(frameBuffer)
|
||||
.rotate()
|
||||
.resize({ width: PREVIEW_MAX_DIMENSION, fit: 'inside', withoutEnlargement: true })
|
||||
.webp({ quality: 85 })
|
||||
.toBuffer();
|
||||
await this.uploadToMinio(previewKey, previewBuffer, 'image/webp');
|
||||
|
||||
await this.photoRepository.update(photoId, {
|
||||
thumbnailKey,
|
||||
previewKey,
|
||||
processingStatus: 'completed',
|
||||
processingError: null,
|
||||
});
|
||||
|
||||
this.logger.logWithData('info', 'Video thumbnail processing completed', {
|
||||
photoId,
|
||||
thumbnailKey,
|
||||
previewKey,
|
||||
});
|
||||
}
|
||||
|
||||
private async extractVideoFrame(storageKey: string, photoId: string): Promise<Buffer | null> {
|
||||
const tmpInput = `/tmp/mg_video_${photoId}`;
|
||||
const tmpOutput = `/tmp/mg_frame_${photoId}.jpg`;
|
||||
|
||||
try {
|
||||
const videoBuffer = await this.fetchFromMinio(storageKey);
|
||||
await fs.writeFile(tmpInput, videoBuffer);
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const child = spawn('ffmpeg', [
|
||||
'-i', tmpInput,
|
||||
'-vframes', '1',
|
||||
'-y',
|
||||
tmpOutput,
|
||||
]);
|
||||
child.on('close', (code) => (code === 0 ? resolve() : reject(new Error(`ffmpeg exited with code ${code}`))));
|
||||
child.on('error', reject);
|
||||
});
|
||||
|
||||
return await fs.readFile(tmpOutput);
|
||||
} catch (error) {
|
||||
this.logger.logWithData('warn', 'Video frame extraction failed', {
|
||||
photoId,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
return null;
|
||||
} finally {
|
||||
await unlinkSilently(tmpInput);
|
||||
await unlinkSilently(tmpOutput);
|
||||
}
|
||||
}
|
||||
|
||||
private async fetchFromMinio(key: string): Promise<Buffer> {
|
||||
return this.minioService.download(key);
|
||||
}
|
||||
|
||||
private async uploadToMinio(key: string, buffer: Buffer, contentType: string): Promise<void> {
|
||||
await this.minioService.uploadBuffer(key, buffer, contentType);
|
||||
}
|
||||
|
||||
private async generateThumbnail(buffer: Buffer, size: number): Promise<Buffer> {
|
||||
return sharp(buffer)
|
||||
.rotate() // Auto-rotate based on EXIF
|
||||
.resize(size, size, {
|
||||
fit: 'cover',
|
||||
position: 'centre',
|
||||
})
|
||||
.webp({ quality: 80 })
|
||||
.toBuffer();
|
||||
}
|
||||
|
||||
private async generatePreview(buffer: Buffer, maxDimension: number, width: number, height: number): Promise<Buffer> {
|
||||
const isLandscape = width > height;
|
||||
const resizeOptions = isLandscape
|
||||
? { width: maxDimension, height: undefined }
|
||||
: { width: undefined, height: maxDimension };
|
||||
|
||||
return sharp(buffer)
|
||||
.rotate() // Auto-rotate based on EXIF
|
||||
.resize({
|
||||
...resizeOptions,
|
||||
fit: 'inside',
|
||||
withoutEnlargement: true,
|
||||
})
|
||||
.webp({ quality: 85 })
|
||||
.toBuffer();
|
||||
}
|
||||
|
||||
private generateThumbnailKey(photoId: string): string {
|
||||
return `thumbnails/${photoId}_thumb.webp`;
|
||||
}
|
||||
|
||||
private generatePreviewKey(photoId: string): string {
|
||||
return `previews/${photoId}_preview.webp`;
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue