diff --git a/apps/api/src/app/app.controller.ts b/apps/api/src/app/app.controller.ts index de999d5..f7ea981 100644 --- a/apps/api/src/app/app.controller.ts +++ b/apps/api/src/app/app.controller.ts @@ -422,8 +422,8 @@ export class AppController { ); } - // Construct MinIO object path for playlist - const objectName = `processed-videos/${videoId}/720p/720p.m3u8`; + // Construct MinIO object path for master playlist + const objectName = `processed-videos/${videoId}/index.m3u8`; try { // Stream the playlist file from MinIO @@ -465,13 +465,12 @@ export class AppController { } } - @Get('videos/:videoId/segments/:segment') - @Header('Content-Type', 'video/mp2t') + @Get('videos/:videoId/:resolution/:filename') @Header('Access-Control-Allow-Origin', '*') @ApiOperation({ - summary: 'Get HLS segment for a video', + summary: 'Get HLS playlist or segment for a video', description: - 'Serves an HLS segment (.ts) file for a processed video. Only available for videos with COMPLETED status.', + 'Serves resolution-specific HLS playlists (.m3u8) or segments (.ts) for a processed video. Handles paths like /videos/:videoId/360p/360p.m3u8 or /videos/:videoId/360p/segment-00001.ts', }) @ApiParam({ name: 'videoId', @@ -480,34 +479,33 @@ export class AppController { example: '123e4567-e89b-12d3-a456-426614174000', }) @ApiParam({ - name: 'segment', - description: 'Segment filename (e.g., segment-00001.ts)', + name: 'resolution', + description: 'Resolution label (e.g., 360p, 720p, 1080p)', type: 'string', - example: 'segment-00001.ts', + example: '360p', }) - @ApiOkResponse({ - description: 'HLS segment file', - content: { - 'video/mp2t': { - schema: { - type: 'string', - format: 'binary', - }, - }, - }, + @ApiParam({ + name: 'filename', + description: 'Playlist (.m3u8) or segment (.ts) filename', + type: 'string', + example: '360p.m3u8', }) - async getSegment( + async getResolutionFile( @Param('videoId') videoId: string, - @Param('segment') segment: string, + @Param('resolution') resolution: string, + @Param('filename') filename: string, @Res() res: Response, ): Promise { - this.logger.log( - `Streaming segment request for video: ${videoId}, segment: ${segment}`, - ); - - // Validate segment filename to prevent path traversal - if (segment.includes('..') || segment.includes('/') || segment.includes('\\')) { - throw new NotFoundException('Invalid segment filename'); + // Validate resolution and filename to prevent path traversal + if ( + resolution.includes('..') || + resolution.includes('/') || + resolution.includes('\\') || + filename.includes('..') || + filename.includes('/') || + filename.includes('\\') + ) { + throw new NotFoundException('Invalid path'); } // Check if video exists @@ -527,21 +525,29 @@ export class AppController { ); } - // Construct MinIO object path for segment - const objectName = `processed-videos/${videoId}/720p/${segment}`; + // Determine content type based on file extension + const isPlaylist = filename.endsWith('.m3u8'); + const contentType = isPlaylist + ? 'application/vnd.apple.mpegurl' + : 'video/mp2t'; + + res.setHeader('Content-Type', contentType); + + // Construct MinIO object path + const objectName = `processed-videos/${videoId}/${resolution}/${filename}`; try { - // Stream the segment file from MinIO + // Stream the file from MinIO const stream = await this.storageService.downloadStream(objectName); stream.pipe(res); stream.on('error', (error) => { this.logger.error( - `Error streaming segment ${segment} for video ${videoId}: ${error.message}`, + `Error streaming ${filename} for video ${videoId}/${resolution}: ${error.message}`, ); if (!res.headersSent) { res.status(HttpStatus.INTERNAL_SERVER_ERROR).json({ - error: 'Failed to stream segment file', + error: `Failed to stream ${isPlaylist ? 'playlist' : 'segment'} file`, message: error.message, }); } @@ -549,12 +555,12 @@ export class AppController { stream.on('end', () => { this.logger.log( - `Segment stream completed for video: ${videoId}, segment: ${segment}`, + `Stream completed for video: ${videoId}/${resolution}/${filename}`, ); }); } catch (error) { this.logger.error( - `Failed to stream segment ${segment} for video ${videoId}: ${error.message}`, + `Failed to stream ${filename} for video ${videoId}/${resolution}: ${error.message}`, ); // Check if it's a not found error @@ -564,7 +570,7 @@ export class AppController { error.message?.includes('NotFound') ) { throw new NotFoundException( - `HLS segment not found: ${segment} for video ${videoId}`, + `${isPlaylist ? 'Playlist' : 'Segment'} not found: ${filename} for video ${videoId}/${resolution}`, ); } // Re-throw other errors (will be handled by NestJS error handler) @@ -572,13 +578,13 @@ export class AppController { } } - @Get('videos/:videoId/:segment') + @Get('videos/:videoId/segments/:segment') @Header('Content-Type', 'video/mp2t') @Header('Access-Control-Allow-Origin', '*') @ApiOperation({ - summary: 'Get HLS segment for a video (direct path)', + summary: 'Get HLS segment for a video (legacy endpoint)', description: - 'Serves an HLS segment (.ts) file directly under the video path. This handles FFmpeg-generated segment names like 720p0.ts, 720p1.ts, etc.', + 'Legacy endpoint for segments. Use /videos/:videoId/:resolution/:filename instead.', }) @ApiParam({ name: 'videoId', @@ -588,9 +594,9 @@ export class AppController { }) @ApiParam({ name: 'segment', - description: 'Segment filename (e.g., 720p0.ts)', + description: 'Segment filename (e.g., segment-00001.ts)', type: 'string', - example: '720p0.ts', + example: 'segment-00001.ts', }) @ApiOkResponse({ description: 'HLS segment file', @@ -603,84 +609,100 @@ export class AppController { }, }, }) - async getSegmentDirect( + async getSegment( @Param('videoId') videoId: string, @Param('segment') segment: string, @Res() res: Response, ): Promise { - // Only handle .ts files to avoid conflicts with other routes - if (!segment.endsWith('.ts')) { - throw new NotFoundException('Invalid segment file'); - } - - this.logger.log( - `Streaming segment (direct) request for video: ${videoId}, segment: ${segment}`, - ); - - // Validate segment filename to prevent path traversal - if (segment.includes('..') || segment.includes('/') || segment.includes('\\')) { - throw new NotFoundException('Invalid segment filename'); + // Legacy endpoint - try 720p first, then other resolutions + const resolutions = ['720p', '360p', '1080p']; + + for (const resolution of resolutions) { + try { + const objectName = `processed-videos/${videoId}/${resolution}/${segment}`; + const stream = await this.storageService.downloadStream(objectName); + stream.pipe(res); + return; + } catch { + // Try next resolution + continue; + } } + throw new NotFoundException(`Segment not found: ${segment} for video ${videoId}`); + } - // Check if video exists - const video = await this.videosService.findOne(videoId); - if (!video) { - this.logger.warn(`Video not found: ${videoId}`); - throw new NotFoundException(`Video with ID ${videoId} not found`); + /** + * Helper method to load HTML files from assets directory. + * Tries multiple possible paths for development and production environments. + */ + private loadHtmlFile(filename: string): string { + const possiblePaths = [ + join(__dirname, '../../assets', filename), // Production (from dist/apps/api/app/app) + join(__dirname, '../assets', filename), // Alternative production path + join(process.cwd(), 'apps/api/src/assets', filename), // Development + join(process.cwd(), 'dist/apps/api/assets', filename), // Production from root + ]; + + for (const htmlPath of possiblePaths) { + try { + const html = readFileSync(htmlPath, 'utf-8'); + this.logger.log(`HTML file loaded from: ${htmlPath}`); + return html; + } catch { + // Try next path + continue; + } } + throw new Error(`Could not find ${filename} in any expected location`); + } - // Check if video processing is completed - if (video.status !== VideoStatus.COMPLETED) { - this.logger.warn( - `Video ${videoId} is not ready for streaming. Status: ${video.status}`, - ); - throw new ConflictException( - `Video processing not completed. Current status: ${video.status}`, - ); + @Get('/') + @Header('Content-Type', 'text/html') + @ApiOperation({ + summary: 'Get main dashboard', + description: 'Serves the main dashboard page with video list and upload functionality.', + }) + @ApiOkResponse({ + description: 'Dashboard HTML page', + content: { + 'text/html': { + schema: { + type: 'string', + }, + }, + }, + }) + getDashboard(): string { + try { + return this.loadHtmlFile('index.html'); + } catch (error) { + this.logger.error(`Failed to load dashboard HTML: ${error.message}`); + throw new Error('Failed to load dashboard page'); } + } - // Construct MinIO object path for segment - const objectName = `processed-videos/${videoId}/720p/${segment}`; - + @Get('video') + @Header('Content-Type', 'text/html') + @ApiOperation({ + summary: 'Get video viewer page', + description: 'Serves the video viewer page with HLS player and video details.', + }) + @ApiOkResponse({ + description: 'Video viewer HTML page', + content: { + 'text/html': { + schema: { + type: 'string', + }, + }, + }, + }) + getVideoViewer(): string { try { - // Stream the segment file from MinIO - const stream = await this.storageService.downloadStream(objectName); - stream.pipe(res); - - stream.on('error', (error) => { - this.logger.error( - `Error streaming segment ${segment} for video ${videoId}: ${error.message}`, - ); - if (!res.headersSent) { - res.status(HttpStatus.INTERNAL_SERVER_ERROR).json({ - error: 'Failed to stream segment file', - message: error.message, - }); - } - }); - - stream.on('end', () => { - this.logger.log( - `Segment stream completed for video: ${videoId}, segment: ${segment}`, - ); - }); + return this.loadHtmlFile('video.html'); } catch (error) { - this.logger.error( - `Failed to stream segment ${segment} for video ${videoId}: ${error.message}`, - ); - - // Check if it's a not found error - if ( - error.message?.includes('not found') || - error.message?.includes('NoSuchKey') || - error.message?.includes('NotFound') - ) { - throw new NotFoundException( - `HLS segment not found: ${segment} for video ${videoId}`, - ); - } - // Re-throw other errors (will be handled by NestJS error handler) - throw error; + this.logger.error(`Failed to load video viewer HTML: ${error.message}`); + throw new Error('Failed to load video viewer page'); } } @@ -703,27 +725,7 @@ export class AppController { }) getTestPlayer(): string { try { - // Try multiple possible paths for the HTML file - // In development: assets are in src/assets - // In production: assets are copied to dist/apps/api/assets - const possiblePaths = [ - join(__dirname, '../../assets/test-player.html'), // Production (from dist/apps/api/app/app) - join(__dirname, '../assets/test-player.html'), // Alternative production path - join(process.cwd(), 'apps/api/src/assets/test-player.html'), // Development - join(process.cwd(), 'dist/apps/api/assets/test-player.html'), // Production from root - ]; - - for (const htmlPath of possiblePaths) { - try { - const html = readFileSync(htmlPath, 'utf-8'); - this.logger.log(`Test player HTML loaded from: ${htmlPath}`); - return html; - } catch { - // Try next path - continue; - } - } - throw new Error('Could not find test-player.html in any expected location'); + return this.loadHtmlFile('test-player.html'); } catch (error) { this.logger.error(`Failed to load test player HTML: ${error.message}`); throw new Error('Failed to load test player page'); diff --git a/apps/api/src/app/queue/queue.service.ts b/apps/api/src/app/queue/queue.service.ts index 33e7b00..1441ff2 100644 --- a/apps/api/src/app/queue/queue.service.ts +++ b/apps/api/src/app/queue/queue.service.ts @@ -2,6 +2,48 @@ import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/commo import { ConfigService } from '@nestjs/config'; import * as amqp from 'amqplib'; +/** + * Configuration for a single video resolution. + */ +export interface ResolutionConfig { + /** Video height in pixels (e.g., 360, 720, 1080) */ + height: number; + + /** Video width in pixels (e.g., 640, 1280, 1920) */ + width: number; + + /** Estimated bandwidth in bits per second (e.g., 800000 for 800 kbps) */ + bandwidth: number; + + /** Resolution label (e.g., '360p', '720p', '1080p') */ + label: string; +} + +/** + * Default resolutions for multi-resolution transcoding. + * Follows 16:9 aspect ratio with standard bandwidth estimates. + */ +export const DEFAULT_RESOLUTIONS: ResolutionConfig[] = [ + { + height: 360, + width: 640, + bandwidth: 800000, // 800 kbps + label: '360p', + }, + { + height: 720, + width: 1280, + bandwidth: 2500000, // 2.5 Mbps + label: '720p', + }, + { + height: 1080, + width: 1920, + bandwidth: 5000000, // 5 Mbps + label: '1080p', + }, +]; + @Injectable() export class QueueService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(QueueService.name); @@ -127,7 +169,8 @@ export class QueueService implements OnModuleInit, OnModuleDestroy { sourceBucket: string, sourceKey: string, targetBucket: string, - profile: 'HLS_720P' = 'HLS_720P', + resolutions?: ResolutionConfig[], + profile?: 'HLS_720P', ): Promise { // Basic input validation if (!videoId || typeof videoId !== 'string') { @@ -150,16 +193,34 @@ export class QueueService implements OnModuleInit, OnModuleDestroy { throw new Error('Invalid targetBucket for publishTranscodeJob.'); } + // Determine which resolutions to use: + // 1. If resolutions is provided, use it + // 2. Otherwise, if profile is provided, use DEFAULT_RESOLUTIONS (backward compatibility) + // 3. Otherwise, use DEFAULT_RESOLUTIONS + const finalResolutions = resolutions || DEFAULT_RESOLUTIONS; + const queueName = this.configService.get('rabbitmq.queueName'); - const jobPayload = { + const jobPayload: { + videoId: string; + sourceBucket: string; + sourceKey: string; + targetBucket: string; + resolutions: ResolutionConfig[]; + profile?: 'HLS_720P'; + } = { videoId, sourceBucket, sourceKey, targetBucket, - profile, + resolutions: finalResolutions, }; + // Include profile for backward compatibility if provided + if (profile) { + jobPayload.profile = profile; + } + const message = JSON.stringify(jobPayload); try { @@ -187,8 +248,9 @@ export class QueueService implements OnModuleInit, OnModuleDestroy { throw new Error(`Failed to publish transcode job for videoId=${videoId}.`); } + const resolutionsLabel = finalResolutions.map((r) => r.label).join(', '); this.logger.log( - `Published transcode job to queue '${queueName}': videoId=${videoId}, sourceBucket=${sourceBucket}, sourceKey=${sourceKey}, targetBucket=${targetBucket}, profile=${profile}`, + `Published transcode job to queue '${queueName}': videoId=${videoId}, sourceBucket=${sourceBucket}, sourceKey=${sourceKey}, targetBucket=${targetBucket}, resolutions=[${resolutionsLabel}]`, ); } catch (error) { this.logger.error( diff --git a/apps/api/src/app/videos/videos.service.ts b/apps/api/src/app/videos/videos.service.ts index 8cda896..8cb2811 100644 --- a/apps/api/src/app/videos/videos.service.ts +++ b/apps/api/src/app/videos/videos.service.ts @@ -4,7 +4,7 @@ import { InjectRepository } from '@nestjs/typeorm'; import { Readable } from 'stream'; import { Repository } from 'typeorm'; import { v4 as uuidv4 } from 'uuid'; -import { QueueService } from '../queue/queue.service'; +import { DEFAULT_RESOLUTIONS, QueueService } from '../queue/queue.service'; import { StorageService } from '../storage/storage.service'; import { Video, VideoStatus } from './entities/video.entity'; @@ -77,10 +77,12 @@ export class VideosService { const sourceKey = s3Key.startsWith(`${sourceBucket}/`) ? s3Key.substring(sourceBucket.length + 1) : s3Key; - const profile = 'HLS_720P' as const; + + const resolutions = DEFAULT_RESOLUTIONS; + const resolutionsLabel = resolutions.map((r) => r.label).join(', '); this.logger.log( - `Publishing transcode job for video: id=${video.id}, sourceBucket=${sourceBucket}, sourceKey=${sourceKey}, targetBucket=${targetBucket}, profile=${profile}`, + `Publishing transcode job for video: id=${video.id}, sourceBucket=${sourceBucket}, sourceKey=${sourceKey}, targetBucket=${targetBucket}, resolutions=[${resolutionsLabel}]`, ); await this.queueService.publishTranscodeJob( @@ -88,7 +90,7 @@ export class VideosService { sourceBucket, sourceKey, targetBucket, - profile, + resolutions, ); // Update status to QUEUED after successful publish diff --git a/apps/api/src/assets/index.html b/apps/api/src/assets/index.html new file mode 100644 index 0000000..8554ab3 --- /dev/null +++ b/apps/api/src/assets/index.html @@ -0,0 +1,499 @@ + + + + + + StreamForge - Video Platform + + + +
+
+

🎬 StreamForge Video Platform

+
0 videos
+
+ +
+
+
+

πŸ“€ Upload Video

+

Drag and drop a video file here, or click to select

+ + +
+
+
+
+
+

Uploading...

+
+
+
+ +
+ +
+
Loading videos...
+
+
+ + + + + diff --git a/apps/api/src/assets/test-player.html b/apps/api/src/assets/test-player.html index bc3d634..84d63b8 100644 --- a/apps/api/src/assets/test-player.html +++ b/apps/api/src/assets/test-player.html @@ -224,7 +224,7 @@

🎬 StreamForge HLS Video Player

πŸ“‹ How to use:

-

1. Get a video ID from GET /api/test-videos endpoint

+

1. Get a video ID from GET /test-videos endpoint

2. Make sure the video status is COMPLETED

3. Enter the video ID above and click "Load Video"

4. The video should start playing automatically

@@ -237,10 +237,10 @@

πŸ“‹ How to use:

let API_BASE_URL; if (window.location.protocol === 'file:' || !window.location.origin || window.location.origin === 'null') { // Default to localhost when opened as file or origin is unavailable - API_BASE_URL = 'http://localhost:3000/api'; + API_BASE_URL = 'http://localhost:3000'; } else { // Use current origin when served from the API - API_BASE_URL = window.location.origin + '/api'; + API_BASE_URL = window.location.origin; } let hls = null; diff --git a/apps/api/src/assets/video.html b/apps/api/src/assets/video.html new file mode 100644 index 0000000..915dda4 --- /dev/null +++ b/apps/api/src/assets/video.html @@ -0,0 +1,485 @@ + + + + + + StreamForge - Video Player + + + + +
+
+ ← Back to Videos +

Loading...

+
+ +
+ +
+ +
Loading video...
+
+ +
+

Video Details

+
+
Loading video information...
+
+
+
+ + + + + diff --git a/apps/api/src/main.ts b/apps/api/src/main.ts index 666b2ad..1312280 100644 --- a/apps/api/src/main.ts +++ b/apps/api/src/main.ts @@ -10,8 +10,6 @@ import { AppModule } from './app/app.module'; async function bootstrap() { const app = await NestFactory.create(AppModule); - const globalPrefix = 'api'; - app.setGlobalPrefix(globalPrefix); // Swagger configuration const config = new DocumentBuilder() @@ -28,7 +26,7 @@ async function bootstrap() { const port = process.env.PORT || 3000; await app.listen(port); Logger.log( - `πŸš€ Application is running on: http://localhost:${port}/${globalPrefix}` + `πŸš€ Application is running on: http://localhost:${port}` ); Logger.log( `πŸ“š Swagger documentation available at: http://localhost:${port}/api/docs` diff --git a/apps/worker/src/app/queue-consumer/queue-consumer.service.ts b/apps/worker/src/app/queue-consumer/queue-consumer.service.ts index 1d55780..be9cde8 100644 --- a/apps/worker/src/app/queue-consumer/queue-consumer.service.ts +++ b/apps/worker/src/app/queue-consumer/queue-consumer.service.ts @@ -2,7 +2,11 @@ import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/commo import { ConfigService } from '@nestjs/config'; import * as amqp from 'amqplib'; import { getFFmpegPath, validateFFmpeg } from '../../utils/ffmpeg.util'; -import { TranscodeJobData } from '../transcoder/dto/transcode-job-data.dto'; +import { + DEFAULT_RESOLUTIONS, + ResolutionConfig, + TranscodeJobData, +} from '../transcoder/dto/transcode-job-data.dto'; import { TranscoderService } from '../transcoder/transcoder.service'; import { VideoStatus } from '../videos/entities/video.entity'; import { VideosService } from '../videos/videos.service'; @@ -190,6 +194,76 @@ export class QueueConsumerService implements OnModuleInit, OnModuleDestroy { } } + /** + * Safely update video status with error handling. + * Wraps status updates in try/catch to prevent failures from breaking the flow. + * + * @param videoId - Video ID (can be null) + * @param status - Status to update to + * @returns Boolean indicating if update was successful + */ + private async updateStatusSafely( + videoId: string | null, + status: VideoStatus, + ): Promise { + if (!videoId) { + this.logger.warn(`Cannot update status to ${status}: videoId is not available`); + return false; + } + + try { + await this.videosService.updateStatus(videoId, status); + this.logger.log(`[${videoId}] Status updated to ${status}`); + return true; + } catch (error) { + this.logger.warn( + `[${videoId}] Failed to update status to ${status}: ${error instanceof Error ? error.message : 'Unknown error'}`, + ); + return false; + } + } + + /** + * Categorize errors that occur before or outside of transcoder service. + * For errors from transcoder service, use the errorInfo attached to the error. + * + * @param error - The error to categorize + * @param videoId - Optional video ID for logging context + * @returns Error info object with category, retryable flag, and phase + */ + private categorizeError(error: Error, videoId?: string): { + category: string; + retryable: boolean; + phase: string; + } { + const errorMessage = error.message || 'Unknown error'; + + // If error has errorInfo from transcoder, use it + if ((error as any).errorInfo) { + return (error as any).errorInfo; + } + + // Validation/parsing errors (non-retryable) + if ( + errorMessage.includes('Missing or invalid') || + errorMessage.includes('JSON') || + errorMessage.includes('Unexpected token') || + errorMessage.includes('cannot be empty') || + errorMessage.includes('Invalid resolution config') + ) { + return { + category: 'VALIDATION_ERROR', + retryable: false, + phase: 'validation', + }; + } // Default to internal error (retryable) + return { + category: 'INTERNAL_ERROR', + retryable: true, + phase: 'unknown', + }; + } + /** * Handle incoming message from the queue. */ @@ -230,8 +304,35 @@ export class QueueConsumerService implements OnModuleInit, OnModuleDestroy { if (!parsedJob.targetBucket || typeof parsedJob.targetBucket !== 'string') { throw new Error('Missing or invalid targetBucket in job payload'); } - if (!parsedJob.profile || parsedJob.profile !== 'HLS_720P') { - throw new Error('Missing or invalid profile in job payload (must be HLS_720P)'); + + // Determine resolutions: prefer resolutions array, fallback to profile, then default + let resolutions: ResolutionConfig[]; + if (parsedJob.resolutions && Array.isArray(parsedJob.resolutions)) { + // Validate resolutions array + if (parsedJob.resolutions.length === 0) { + throw new Error('Resolutions array cannot be empty'); + } + // Validate each resolution config + for (const res of parsedJob.resolutions) { + if ( + !res || + typeof res.height !== 'number' || + typeof res.width !== 'number' || + typeof res.bandwidth !== 'number' || + typeof res.label !== 'string' + ) { + throw new Error( + 'Invalid resolution config: each resolution must have height, width, bandwidth (numbers) and label (string)', + ); + } + } + resolutions = parsedJob.resolutions; + } else if (parsedJob.profile === 'HLS_720P') { + // Backward compatibility: convert profile to default resolutions + resolutions = DEFAULT_RESOLUTIONS; + } else { + // Default to standard resolutions if neither is provided + resolutions = DEFAULT_RESOLUTIONS; } const job: TranscodeJobData = { @@ -239,38 +340,31 @@ export class QueueConsumerService implements OnModuleInit, OnModuleDestroy { sourceBucket: parsedJob.sourceBucket, sourceKey: parsedJob.sourceKey, targetBucket: parsedJob.targetBucket, - profile: parsedJob.profile, + resolutions, }; + // Include profile for backward compatibility if provided + if (parsedJob.profile) { + job.profile = parsedJob.profile; + } + videoId = job.videoId; jobId = `job-${Date.now()}-${Math.random().toString(36).substring(7)}`; this.inFlightJobs.add(jobId); + const resolutionsLabel = resolutions.map((r) => r.label).join(', '); this.logger.log( - `[${videoId}] Received transcode job ${jobId}: sourceBucket=${job.sourceBucket}, sourceKey=${job.sourceKey}, targetBucket=${job.targetBucket}, profile=${job.profile}`, + `[${videoId}] Received transcode job ${jobId}: sourceBucket=${job.sourceBucket}, sourceKey=${job.sourceKey}, targetBucket=${job.targetBucket}, resolutions=[${resolutionsLabel}]`, ); // Update video status to PROCESSING - try { - await this.videosService.updateStatus(videoId, VideoStatus.PROCESSING); - } catch (statusError) { - this.logger.warn( - `[${videoId}] Failed to update status to PROCESSING (continuing anyway): ${statusError instanceof Error ? statusError.message : 'Unknown error'}`, - ); - } + await this.updateStatusSafely(videoId, VideoStatus.PROCESSING); // Process the transcoding job await this.transcoderService.transcodeVideo(job); // Update video status to COMPLETED on success - try { - await this.videosService.updateStatus(videoId, VideoStatus.COMPLETED); - this.logger.log(`[${videoId}] Video status updated to COMPLETED`); - } catch (statusError) { - this.logger.warn( - `[${videoId}] Failed to update status to COMPLETED: ${statusError instanceof Error ? statusError.message : 'Unknown error'}`, - ); - } + await this.updateStatusSafely(videoId, VideoStatus.COMPLETED); // Acknowledge message on success try { @@ -284,60 +378,58 @@ export class QueueConsumerService implements OnModuleInit, OnModuleDestroy { const errorMessage = error instanceof Error ? error.message : 'Unknown error'; const errorStack = error instanceof Error ? error.stack : undefined; - // Check if this is a parsing/validation error (non-retryable) - if ( - errorMessage.includes('Missing or invalid') || - errorMessage.includes('JSON') || - errorMessage.includes('Unexpected token') - ) { + // Categorize the error + const errorInfo = this.categorizeError(error as Error, videoId || undefined); + const isRetryable = errorInfo.retryable; + const retryCount = (msg.properties.headers?.['x-retry-count'] as number) || 0; + const maxRetries = this.configService.get('transcode.maxRetries') || 1; + + // Comprehensive error logging with full context + this.logger.error( + `[${videoId || 'unknown'}] Job ${jobId || 'unknown'} failed - ` + + `Category: ${errorInfo.category}, ` + + `Phase: ${errorInfo.phase}, ` + + `Retryable: ${isRetryable}, ` + + `Retry count: ${retryCount}/${maxRetries}, ` + + `Error: ${errorMessage}`, + errorStack, + ); + + // Log message properties for debugging + if (msg.properties) { + this.logger.debug( + `[${videoId || 'unknown'}] Message properties: ${JSON.stringify(msg.properties)}`, + ); + } + + // Handle validation errors (non-retryable) + if (errorInfo.category === 'VALIDATION_ERROR') { this.logger.error( `[${videoId || 'unknown'}] Malformed job message (non-retryable): ${errorMessage}`, - errorStack, ); + // Update status to FAILED if videoId is available + await this.updateStatusSafely(videoId, VideoStatus.FAILED); // Nack without requeue for malformed messages this.channel.nack(msg, false, false); - if (videoId) { - try { - await this.videosService.updateStatus(videoId, VideoStatus.FAILED); - } catch (statusError) { - this.logger.warn( - `[${videoId}] Failed to update status to FAILED: ${statusError instanceof Error ? statusError.message : 'Unknown error'}`, - ); - } - } return; } - // For transcoding errors, check if retryable + // For errors without videoId, we can't update status but should still handle the message if (!videoId) { - this.logger.error('Error occurred but videoId is not available, nacking without requeue'); + this.logger.error( + `Error occurred but videoId is not available, nacking without requeue. Error: ${errorMessage}`, + ); this.channel.nack(msg, false, false); return; } - // Get error categorization from the error object (attached by TranscoderService) - const errorInfo = (error as any).errorInfo; - const isRetryable = errorInfo?.retryable ?? true; // Default to retryable if not available - const retryCount = (msg.properties.headers?.['x-retry-count'] as number) || 0; - const maxRetries = this.configService.get('transcode.maxRetries') || 3; - - this.logger.error( - `[${videoId}] Job ${jobId} failed - Category: ${errorInfo?.category || 'UNKNOWN'}, Retryable: ${isRetryable}, Retry count: ${retryCount}/${maxRetries}`, - errorStack, - ); - + // For transcoding and other errors, check retry logic if (!isRetryable || retryCount >= maxRetries) { // Non-retryable error or max retries exceeded this.logger.error( `[${videoId}] Job ${jobId} failed permanently - updating status to FAILED`, ); - try { - await this.videosService.updateStatus(videoId, VideoStatus.FAILED); - } catch (statusError) { - this.logger.warn( - `[${videoId}] Failed to update status to FAILED: ${statusError instanceof Error ? statusError.message : 'Unknown error'}`, - ); - } + await this.updateStatusSafely(videoId, VideoStatus.FAILED); // Nack without requeue (dead-letter) this.channel.nack(msg, false, false); } else { @@ -354,7 +446,11 @@ export class QueueConsumerService implements OnModuleInit, OnModuleDestroy { } } finally { if (jobId) { + this.logger.debug(`[${videoId || 'unknown'}] Removing job ${jobId} from in-flight jobs`); this.inFlightJobs.delete(jobId); + this.logger.debug( + `[${videoId || 'unknown'}] Remaining in-flight jobs: ${this.inFlightJobs.size}`, + ); } } } diff --git a/apps/worker/src/app/transcoder/dto/transcode-job-data.dto.ts b/apps/worker/src/app/transcoder/dto/transcode-job-data.dto.ts index 22b82dc..d38e856 100644 --- a/apps/worker/src/app/transcoder/dto/transcode-job-data.dto.ts +++ b/apps/worker/src/app/transcoder/dto/transcode-job-data.dto.ts @@ -1,3 +1,45 @@ +/** + * Configuration for a single video resolution. + */ +export interface ResolutionConfig { + /** Video height in pixels (e.g., 360, 720, 1080) */ + height: number; + + /** Video width in pixels (e.g., 640, 1280, 1920) */ + width: number; + + /** Estimated bandwidth in bits per second (e.g., 800000 for 800 kbps) */ + bandwidth: number; + + /** Resolution label (e.g., '360p', '720p', '1080p') */ + label: string; +} + +/** + * Default resolutions for multi-resolution transcoding. + * Follows 16:9 aspect ratio with standard bandwidth estimates. + */ +export const DEFAULT_RESOLUTIONS: ResolutionConfig[] = [ + { + height: 360, + width: 640, + bandwidth: 800000, // 800 kbps + label: '360p', + }, + { + height: 720, + width: 1280, + bandwidth: 2500000, // 2.5 Mbps + label: '720p', + }, + { + height: 1080, + width: 1920, + bandwidth: 5000000, // 5 Mbps + label: '1080p', + }, +]; + /** * Data structure for a transcoding job. * Matches the PRD specification for transcode job payload. @@ -15,7 +57,17 @@ export interface TranscodeJobData { /** MinIO bucket where processed HLS files will be stored */ targetBucket: string; - /** Transcoding profile (currently only HLS_720P is supported) */ - profile: 'HLS_720P'; + /** + * Array of resolutions to transcode. + * If not provided, defaults to DEFAULT_RESOLUTIONS (360p, 720p, 1080p). + */ + resolutions?: ResolutionConfig[]; + + /** + * Transcoding profile (deprecated, kept for backward compatibility). + * @deprecated Use `resolutions` array instead. If `resolutions` is not provided, + * this field will be used to determine default resolutions. + */ + profile?: 'HLS_720P'; } diff --git a/apps/worker/src/app/transcoder/transcoder.service.ts b/apps/worker/src/app/transcoder/transcoder.service.ts index 5730176..a1812af 100644 --- a/apps/worker/src/app/transcoder/transcoder.service.ts +++ b/apps/worker/src/app/transcoder/transcoder.service.ts @@ -12,7 +12,11 @@ import { StorageOperationError, } from '../storage/exceptions/storage.exceptions'; import { StorageService } from '../storage/storage.service'; -import { TranscodeJobData } from './dto/transcode-job-data.dto'; +import { + DEFAULT_RESOLUTIONS, + ResolutionConfig, + TranscodeJobData, +} from './dto/transcode-job-data.dto'; /** * Error categories for transcoding operations @@ -35,10 +39,10 @@ export class TranscoderService { ) {} /** - * Transcode a video from source format to 720p HLS. - * Orchestrates the full lifecycle: download, transcode, upload, cleanup. + * Transcode a video to multiple resolutions (360p, 720p, 1080p) in HLS format. + * Orchestrates the full lifecycle: download, transcode each resolution, upload, cleanup. * - * @param job - Transcode job data containing video ID, source/target locations, and profile + * @param job - Transcode job data containing video ID, source/target locations, and resolutions * @returns Promise resolving when transcoding is complete * @throws Error categorized by phase (download/transcode/upload/internal) */ @@ -46,11 +50,17 @@ export class TranscoderService { const jobId = `job-${Date.now()}-${Math.random().toString(36).substring(7)}`; const { videoId, sourceBucket, sourceKey, targetBucket } = job; + // Get resolutions from job or use defaults + const resolutions = job.resolutions || DEFAULT_RESOLUTIONS; + const resolutionsLabel = resolutions.map((r) => r.label).join(', '); + this.logger.log( - `[${videoId}] Starting transcoding job ${jobId} - profile: ${job.profile}`, + `[${videoId}] Starting transcoding job ${jobId} - resolutions: [${resolutionsLabel}]`, ); let tempDir: string | null = null; + const successfulResolutions: string[] = []; + const failedResolutions: Array<{ label: string; error: string }> = []; try { // Set status to PROCESSING and report 0% progress @@ -84,51 +94,214 @@ export class TranscoderService { // Report 10% progress after download completes await this.reportProgress(videoId, 10); - // Step 3: Create output directory - const outputDir = path.join(tempDir, 'output'); - await fs.ensureDir(outputDir); - this.logger.log(`[${videoId}] Output directory created: ${outputDir}`); + // Step 3: Create base output directory + const baseOutputDir = path.join(tempDir, 'output'); + await fs.ensureDir(baseOutputDir); + this.logger.log(`[${videoId}] Base output directory created: ${baseOutputDir}`); + + // Step 4: Transcode each resolution sequentially + const numResolutions = resolutions.length; + const transcodeProgressStart = 10; // Start at 10% (after download) + const transcodeProgressEnd = 70; // End at 70% (before upload) + + for (let i = 0; i < resolutions.length; i++) { + const resolution = resolutions[i]; + let resolutionStartProgress: number; + let resolutionEndProgress: number; + + if (numResolutions === 3) { + // Exact PRD milestones for 3 resolutions + const milestones = [ + { start: 10, complete: 20 }, // 360p + { start: 20, complete: 40 }, // 720p + { start: 40, complete: 70 }, // 1080p + ]; + resolutionStartProgress = milestones[i].start; + resolutionEndProgress = milestones[i].complete; + } else { + // Dynamic calculation for other numbers of resolutions + const transcodeProgressRange = transcodeProgressEnd - transcodeProgressStart; + const progressPerResolution = transcodeProgressRange / numResolutions; + resolutionStartProgress = + transcodeProgressStart + i * progressPerResolution; + resolutionEndProgress = + transcodeProgressStart + (i + 1) * progressPerResolution; + } - // Step 4: Execute FFmpeg transcoding - const transcodeStartTime = Date.now(); - const outputPath = path.join(outputDir, '720p.m3u8'); + try { + this.logger.log( + `[${videoId}] Starting transcoding for ${resolution.label} (${resolution.width}x${resolution.height})...`, + ); - this.logger.log( - `[${videoId}] Starting FFmpeg transcoding to 720p HLS...`, - ); + // Create resolution-specific output directory + const resolutionOutputDir = path.join(baseOutputDir, resolution.label); + await fs.ensureDir(resolutionOutputDir); + const outputPath = path.join(resolutionOutputDir, `${resolution.label}.m3u8`); + + // Report progress when transcoding starts for this resolution + await this.reportProgress(videoId, resolutionStartProgress); + + // Execute FFmpeg transcoding for this resolution + const transcodeStartTime = Date.now(); + await this.executeFFmpeg( + inputPath, + outputPath, + videoId, + resolution, + resolutionStartProgress, + resolutionEndProgress, + ); - await this.executeFFmpeg(inputPath, outputPath, videoId); + const transcodeDuration = Date.now() - transcodeStartTime; + this.logger.log( + `[${videoId}] ${resolution.label} transcoding completed successfully (${transcodeDuration}ms)`, + ); - const transcodeDuration = Date.now() - transcodeStartTime; - this.logger.log( - `[${videoId}] FFmpeg transcoding completed successfully (${transcodeDuration}ms)`, - ); + // Report progress when transcoding completes for this resolution + await this.reportProgress(videoId, resolutionEndProgress); + + // Calculate upload progress milestones + // For 3 resolutions, use exact PRD milestones: 70%, 80%, 85%, 90% + // For other numbers, calculate dynamically + const uploadProgressStart = 70; + const uploadProgressEnd = 95; + let uploadStartProgress: number; + let uploadCompleteProgress: number; + + if (numResolutions === 3) { + // Exact PRD milestones for 3 resolutions + const milestones = [ + { start: 70, complete: 80 }, // 360p + { start: 80, complete: 85 }, // 720p + { start: 85, complete: 90 }, // 1080p + ]; + uploadStartProgress = milestones[i].start; + uploadCompleteProgress = milestones[i].complete; + } else { + // Dynamic calculation for other numbers of resolutions + const uploadProgressRange = uploadProgressEnd - uploadProgressStart; + const uploadProgressPerResolution = uploadProgressRange / numResolutions; + uploadStartProgress = uploadProgressStart + i * uploadProgressPerResolution; + uploadCompleteProgress = + uploadProgressStart + (i + 1) * uploadProgressPerResolution; + } + + // Report progress when upload starts + await this.reportProgress(videoId, uploadStartProgress); + + // Upload this resolution's HLS output to MinIO + const uploadStartTime = Date.now(); + const baseKey = `${videoId}/${resolution.label}`; + + this.logger.log( + `[${videoId}] Uploading ${resolution.label} HLS output to ${targetBucket}/${baseKey}...`, + ); + + await this.storageService.uploadDirectory( + targetBucket, + baseKey, + resolutionOutputDir, + ); + + const uploadDuration = Date.now() - uploadStartTime; + this.logger.log( + `[${videoId}] ${resolution.label} HLS output uploaded successfully (${uploadDuration}ms)`, + ); + + successfulResolutions.push(resolution.label); + + // Report progress when upload completes + await this.reportProgress(videoId, uploadCompleteProgress); + } catch (error) { + const errorMessage = (error as Error).message || 'Unknown error'; + this.logger.error( + `[${videoId}] Failed to transcode ${resolution.label}: ${errorMessage}`, + (error as Error).stack, + ); + failedResolutions.push({ label: resolution.label, error: errorMessage }); + // Continue to next resolution even if this one failed + } + } - // Report 90% progress after FFmpeg completes + // Check if all resolutions failed + if (successfulResolutions.length === 0) { + const errorMessage = `All resolutions failed: ${failedResolutions.map((f) => `${f.label} (${f.error})`).join(', ')}`; + throw new Error(errorMessage); + } + + // Log summary + if (failedResolutions.length > 0) { + this.logger.warn( + `[${videoId}] Some resolutions failed: ${failedResolutions.map((f) => f.label).join(', ')}. Successful: ${successfulResolutions.join(', ')}`, + ); + } + + // Report 90% progress after all resolution uploads complete await this.reportProgress(videoId, 90); - // Step 5: Upload HLS output to MinIO - const uploadStartTime = Date.now(); - const baseKey = `${videoId}/720p`; + // Step 5: Generate and upload master playlist + try { + this.logger.log( + `[${videoId}] Generating master playlist for successful resolutions: ${successfulResolutions.join(', ')}`, + ); - this.logger.log( - `[${videoId}] Uploading HLS output to ${targetBucket}/${baseKey}...`, - ); + const masterPlaylistContent = this.generateMasterPlaylist( + successfulResolutions, + resolutions, + ); - await this.storageService.uploadDirectory(targetBucket, baseKey, outputDir); + // Write master playlist to temporary file + const masterPlaylistPath = path.join(baseOutputDir, 'index.m3u8'); + await fs.writeFile(masterPlaylistPath, masterPlaylistContent, 'utf8'); - const uploadDuration = Date.now() - uploadStartTime; - this.logger.log( - `[${videoId}] HLS output uploaded successfully (${uploadDuration}ms)`, - ); + this.logger.log( + `[${videoId}] Master playlist generated: ${masterPlaylistPath}`, + ); - // Report 100% progress and COMPLETED status after upload completes + // Report 95% progress after master playlist generation + await this.reportProgress(videoId, 95); + + // Upload master playlist to MinIO + const masterPlaylistKey = `${videoId}/index.m3u8`; + const masterPlaylistMetadata = { + 'Content-Type': 'application/vnd.apple.mpegurl', + }; + + this.logger.log( + `[${videoId}] Uploading master playlist to ${targetBucket}/${masterPlaylistKey}...`, + ); + + const masterPlaylistUploadStartTime = Date.now(); + await this.storageService.uploadFile( + targetBucket, + masterPlaylistKey, + masterPlaylistPath, + masterPlaylistMetadata, + ); + + const masterPlaylistUploadDuration = + Date.now() - masterPlaylistUploadStartTime; + this.logger.log( + `[${videoId}] Master playlist uploaded successfully (${masterPlaylistUploadDuration}ms)`, + ); + } catch (masterPlaylistError) { + // Log error but don't fail the job - master playlist is optional + this.logger.error( + `[${videoId}] Failed to generate/upload master playlist: ${(masterPlaylistError as Error).message}`, + (masterPlaylistError as Error).stack, + ); + this.logger.warn( + `[${videoId}] Continuing without master playlist - job will still be marked as completed`, + ); + } + + // Report 100% progress and COMPLETED status await this.reportProgress(videoId, 100); await this.redisService.setJobStatus(videoId, JobStatus.COMPLETED); const totalDuration = Date.now() - tempDirStartTime; this.logger.log( - `[${videoId}] Transcoding job ${jobId} completed successfully in ${totalDuration}ms`, + `[${videoId}] Transcoding job ${jobId} completed successfully in ${totalDuration}ms. Successful resolutions: ${successfulResolutions.join(', ')}`, ); } catch (error) { const errorInfo = this.categorizeError(error as Error, videoId); @@ -178,11 +351,43 @@ export class TranscoderService { } /** - * Execute FFmpeg command to transcode video to 720p HLS. + * Generate HLS master playlist content that references all successful resolutions. + * + * @param successfulResolutions - Array of resolution labels that were successfully transcoded + * @param allResolutions - Array of all resolution configurations + * @returns Master playlist content as a string + */ + private generateMasterPlaylist( + successfulResolutions: string[], + allResolutions: ResolutionConfig[], + ): string { + // Filter to only successful resolutions + const successful = allResolutions.filter((r) => + successfulResolutions.includes(r.label), + ); + + // Sort by bandwidth (lowest to highest) for proper adaptive streaming + successful.sort((a, b) => a.bandwidth - b.bandwidth); + + // Generate playlist content + let playlist = '#EXTM3U\n#EXT-X-VERSION:3\n'; + + for (const resolution of successful) { + playlist += `#EXT-X-STREAM-INF:BANDWIDTH=${resolution.bandwidth},RESOLUTION=${resolution.width}x${resolution.height}\n`; + playlist += `${resolution.label}/${resolution.label}.m3u8\n`; + } + return playlist; + } + + /** + * Execute FFmpeg command to transcode video to a specific resolution in HLS format. * * @param inputPath - Path to input video file - * @param outputPath - Path to output HLS playlist file (720p.m3u8) + * @param outputPath - Path to output HLS playlist file (e.g., 360p.m3u8) * @param videoId - Video ID for logging + * @param resolution - Resolution configuration (width, height, bandwidth, label) + * @param progressStart - Starting progress percentage for this resolution + * @param progressEnd - Ending progress percentage for this resolution * @returns Promise resolving when FFmpeg completes successfully * @throws Error if FFmpeg fails */ @@ -190,50 +395,77 @@ export class TranscoderService { inputPath: string, outputPath: string, videoId: string, + resolution: ResolutionConfig, + progressStart: number, + progressEnd: number, ): Promise { return new Promise((resolve, reject) => { const ffmpegPath = getFFmpegPath(); let stderrOutput = ''; let commandLine = ''; - let lastReportedProgress = 0; + let lastReportedProgress = progressStart; let lastProgressReportTime = Date.now(); - const PROGRESS_THROTTLE_PERCENT = 5; // Report every 5% change + const PROGRESS_THROTTLE_PERCENT = 2; // Report every 2% change const PROGRESS_THROTTLE_MS = 5000; // Or every 5 seconds + // Calculate bitrate based on resolution (bandwidth in bits per second) + // Convert to kbps for FFmpeg (divide by 1000) + const videoBitrate = Math.floor(resolution.bandwidth / 1000); + // Audio bitrate is typically 128 kbps for all resolutions + const audioBitrate = 128; + + // Build output options based on resolution + const outputOptions: string[] = [ + '-g 48', + '-keyint_min 48', + '-sc_threshold 0', + '-hls_time 4', + '-hls_playlist_type vod', + `-b:v ${videoBitrate}k`, // Video bitrate + `-b:a ${audioBitrate}k`, // Audio bitrate + ]; + + // Add quality presets based on resolution + if (resolution.height <= 360) { + // Lower quality preset for 360p (faster encoding) + outputOptions.push('-preset fast'); + } else if (resolution.height >= 1080) { + // Higher quality preset for 1080p + outputOptions.push('-preset medium'); + } else { + // Default preset for 720p + outputOptions.push('-preset fast'); + } + const command = ffmpeg(inputPath) .setFfmpegPath(ffmpegPath) .videoCodec('libx264') .audioCodec('aac') - .size('1280x720') - .outputOptions([ - '-g 48', - '-keyint_min 48', - '-sc_threshold 0', - '-hls_time 4', - '-hls_playlist_type vod', - ]) + .size(`${resolution.width}x${resolution.height}`) + .outputOptions(outputOptions) .output(outputPath); command .on('start', async (cmdline) => { commandLine = cmdline; - this.logger.log(`[${videoId}] FFmpeg command: ${cmdline}`); - // Report 20% progress when FFmpeg starts - await this.reportProgress(videoId, 20); + this.logger.log( + `[${videoId}] FFmpeg command for ${resolution.label}: ${cmdline}`, + ); }) .on('progress', async (progress) => { // Log progress periodically if (progress.percent !== undefined) { const ffmpegPercent = progress.percent; this.logger.debug( - `[${videoId}] FFmpeg progress: ${Math.round(ffmpegPercent)}%`, + `[${videoId}] FFmpeg progress for ${resolution.label}: ${Math.round(ffmpegPercent)}%`, ); - // Map FFmpeg progress (0-100%) to overall progress (20-90%) - // Formula: overallProgress = 20 + (ffmpegProgress * 0.7) - const overallProgress = 20 + ffmpegPercent * 0.7; + // Map FFmpeg progress (0-100%) to resolution-specific progress range + // Formula: overallProgress = progressStart + (ffmpegProgress * (progressEnd - progressStart) / 100) + const progressRange = progressEnd - progressStart; + const overallProgress = progressStart + (ffmpegPercent * progressRange) / 100; - // Throttle progress updates: report every 5% change or every 5 seconds + // Throttle progress updates: report every 2% change or every 5 seconds const now = Date.now(); const progressChange = Math.abs(overallProgress - lastReportedProgress); const timeSinceLastReport = now - lastProgressReportTime; @@ -252,20 +484,24 @@ export class TranscoderService { stderrOutput += stderrLine + '\n'; }) .on('end', () => { - this.logger.log(`[${videoId}] FFmpeg transcoding completed`); + this.logger.log( + `[${videoId}] FFmpeg transcoding completed for ${resolution.label}`, + ); resolve(); }) .on('error', (error) => { this.logger.error( - `[${videoId}] FFmpeg error: ${error.message}`, + `[${videoId}] FFmpeg error for ${resolution.label}: ${error.message}`, error.stack, ); if (stderrOutput) { - this.logger.error(`[${videoId}] FFmpeg stderr output:\n${stderrOutput}`); + this.logger.error( + `[${videoId}] FFmpeg stderr output for ${resolution.label}:\n${stderrOutput}`, + ); } // Create a more descriptive error const ffmpegError = new Error( - `FFmpeg transcoding failed: ${error.message}. Command: ${commandLine}`, + `FFmpeg transcoding failed for ${resolution.label}: ${error.message}. Command: ${commandLine}`, ); (ffmpegError as any).message = stderrOutput; reject(ffmpegError); diff --git a/docs/active/PRD-PROJECT-COMPLETION.md b/docs/active/PRD-PROJECT-COMPLETION.md new file mode 100644 index 0000000..745a1c4 --- /dev/null +++ b/docs/active/PRD-PROJECT-COMPLETION.md @@ -0,0 +1,972 @@ +# Product Requirements Document: StreamForge Project Completion + +## Executive Summary + +This document defines the requirements for completing the StreamForge video transcoding platform, focusing on testing infrastructure, production readiness, security enhancements, and stretch goals. The StreamForge core functionality is complete and operational, with all primary features implemented including multi-resolution transcoding, progress tracking, and error handling. This PRD addresses the remaining gaps to make the platform production-ready and maintainable. + +At a high level, the implementation will be completed in the following phases: + +1. **Phase 1: Testing Infrastructure** + Implement comprehensive unit tests and integration tests for all core services to ensure reliability and catch regressions. + +2. **Phase 2: Production Readiness** + Dockerize applications, add health checks, improve logging, and create production deployment configurations. + +3. **Phase 3: Security & Performance** + Add rate limiting, input validation, CORS configuration, and performance optimizations. + +4. **Phase 4: Stretch Goals** + Implement job cancellation, video metadata extraction, and enhanced API endpoints. + +5. **Phase 5: Documentation & Polish** + Enhance API documentation, README, and code documentation. + +**Timeline**: 4 weeks (estimated 40-60 hours) +**Effort Estimate**: 40-60 hours +**Priority**: High (Critical for production deployment and maintainability) + +--- + +## Background & Context + +### Current Project Status + +StreamForge is functionally complete with all core features implemented: + +βœ… **Completed Features:** +- Infrastructure: Docker Compose with PostgreSQL, Redis, RabbitMQ, MinIO +- MinIO Storage Service: Streaming upload/download with comprehensive error handling +- Database: TypeORM with Video entity and status tracking +- RabbitMQ Integration: Queue service and worker consumer with retry logic +- FFmpeg Transcoding: Multi-resolution (360p, 720p, 1080p) HLS transcoding +- Redis Progress Tracking: Real-time progress updates with TTL +- API Endpoints: Upload, progress tracking, playlist serving, HTML dashboard +- Error Handling: Comprehensive error categorization and retry logic +- Master Playlist: Adaptive bitrate streaming support + +### Why This Completion Phase is Needed + +While StreamForge is functionally complete, several critical gaps prevent it from being production-ready: + +1. **Testing Gap**: No unit tests exist for core services, making it difficult to catch regressions and verify behavior +2. **Production Deployment**: Applications are not containerized, making deployment complex +3. **Observability**: No health checks or monitoring endpoints for production monitoring +4. **Security**: Missing rate limiting, input validation, and proper CORS configuration +5. **Documentation**: API documentation and deployment guides are incomplete + +### Integration with Existing System + +This completion phase builds upon: +- **Storage Service** (`apps/api/src/app/storage/storage.service.ts`): Will add unit tests +- **Queue Service** (`apps/api/src/app/queue/queue.service.ts`): Will add unit tests and health checks +- **Redis Service** (`apps/api/src/app/redis/redis.service.ts`): Will add unit tests +- **Transcoder Service** (`apps/worker/src/app/transcoder/transcoder.service.ts`): Will add unit tests +- **Queue Consumer** (`apps/worker/src/app/queue-consumer/queue-consumer.service.ts`): Will add unit tests +- **API Controller** (`apps/api/src/app/app.controller.ts`): Will add health checks and enhanced endpoints +- **Configuration** (`apps/api/src/config/configuration.ts`): Will add environment validation + +### Current State + +- Core functionality is working and tested manually +- Error handling and retry logic are well-implemented +- Multi-resolution transcoding and progress tracking are production-ready +- Basic e2e tests exist but are minimal +- No unit tests for services +- No Dockerfiles for applications +- No health check endpoints +- No rate limiting or input validation + +### Missing Components + +- Unit test files for all core services +- Integration tests for end-to-end flows +- Dockerfiles for API and Worker applications +- Health check endpoints +- Rate limiting middleware +- Enhanced input validation +- Job cancellation functionality +- Video metadata extraction +- Enhanced API documentation + +--- + +## Objectives & Goals + +### Primary Objectives + +1. **Comprehensive Testing**: Achieve 70%+ code coverage on critical services with unit and integration tests +2. **Production Deployment**: Dockerize applications and create production-ready configurations +3. **Observability**: Add health checks and monitoring endpoints for production monitoring +4. **Security**: Implement rate limiting, input validation, and proper CORS configuration +5. **Enhanced Features**: Add job cancellation and video metadata extraction +6. **Documentation**: Complete API documentation and deployment guides + +### Success Criteria + +- βœ… Unit tests exist for all core services (Storage, Queue, Redis, Transcoder, QueueConsumer) +- βœ… Integration tests cover end-to-end upload β†’ transcode β†’ progress β†’ playlist flow +- βœ… Test coverage is 70%+ on critical services +- βœ… Dockerfiles exist for API and Worker applications +- βœ… Health check endpoint returns status of all dependencies +- βœ… Rate limiting is configured and working +- βœ… Input validation prevents invalid uploads +- βœ… Job cancellation endpoint works correctly +- βœ… Video metadata extraction works and is stored +- βœ… API documentation is complete with examples +- βœ… README includes deployment guide and architecture diagram + +### Key Deliverables + +- Unit test files for all core services +- Enhanced integration tests +- Dockerfiles for API and Worker +- Health check endpoint implementation +- Rate limiting configuration +- Enhanced input validation +- Job cancellation endpoint +- Video metadata extraction +- Enhanced API documentation +- Updated README with deployment guide + +--- + +## Functional Requirements + +### Phase 1: Testing Infrastructure + +#### FR1.1: Storage Service Unit Tests + +**Requirement**: Create comprehensive unit tests for StorageService. + +**Details**: +- Mock MinIO client using Jest mocks +- Test connection initialization +- Test bucket creation logic (including idempotent behavior) +- Test `uploadStream()` method with various stream types +- Test `downloadStream()` method +- Test `objectExists()` method +- Test `deleteObject()` method (including idempotent behavior) +- Test error handling scenarios (connection errors, not found errors, validation errors) +- Test with various object names and bucket configurations + +**Test File**: `apps/api/src/app/storage/storage.service.spec.ts` + +**Test Cases**: +- Connection initialization succeeds +- Connection initialization fails gracefully +- Bucket creation succeeds when bucket doesn't exist +- Bucket creation is idempotent (handles existing bucket) +- Upload stream succeeds with valid stream +- Upload stream fails with invalid stream +- Download stream succeeds for existing object +- Download stream fails for non-existent object +- Object exists returns true for existing object +- Object exists returns false for non-existent object +- Delete object succeeds +- Delete object is idempotent (doesn't fail if already deleted) +- Error handling throws appropriate custom exceptions + +**Acceptance Criteria**: +- All test cases pass +- Code coverage is 80%+ for StorageService +- Tests use proper mocking (no real MinIO connection) +- Tests are fast (< 1 second total) + +--- + +#### FR1.2: Queue Service Unit Tests + +**Requirement**: Create comprehensive unit tests for QueueService. + +**Details**: +- Mock amqplib connection and channel +- Test connection initialization +- Test queue declaration logic +- Test `publishTranscodeJob()` method with various job parameters +- Test error handling scenarios (connection failures, publish failures) +- Test message serialization (JSON format) +- Test persistent message flag + +**Test File**: `apps/api/src/app/queue/queue.service.spec.ts` + +**Test Cases**: +- Connection initialization succeeds +- Connection initialization fails gracefully +- Queue declaration succeeds +- Queue declaration is idempotent +- Publish job succeeds with valid parameters +- Publish job fails with invalid parameters +- Publish job serializes message correctly +- Publish job sets persistent flag +- Error handling throws appropriate errors + +**Acceptance Criteria**: +- All test cases pass +- Code coverage is 80%+ for QueueService +- Tests use proper mocking (no real RabbitMQ connection) +- Tests are fast (< 1 second total) + +--- + +#### FR1.3: Redis Service Unit Tests + +**Requirement**: Create comprehensive unit tests for RedisService. + +**Details**: +- Mock ioredis client +- Test connection initialization +- Test `setProgress()` method +- Test `getProgress()` method +- Test `setJobStatus()` method +- Test TTL expiration (24 hours) +- Test error handling scenarios (connection failures) +- Test JSON serialization/deserialization + +**Test File**: `apps/api/src/app/redis/redis.service.spec.ts` + +**Test Cases**: +- Connection initialization succeeds +- Connection initialization fails gracefully +- Set progress stores data correctly +- Get progress retrieves data correctly +- Get progress returns null for non-existent key +- Set job status stores status and error correctly +- TTL is set correctly (24 hours) +- Error handling handles connection failures gracefully +- JSON serialization/deserialization works correctly + +**Acceptance Criteria**: +- All test cases pass +- Code coverage is 80%+ for RedisService +- Tests use proper mocking (no real Redis connection) +- Tests are fast (< 1 second total) + +--- + +#### FR1.4: Transcoder Service Unit Tests + +**Requirement**: Create comprehensive unit tests for TranscoderService. + +**Details**: +- Mock FFmpeg (fluent-ffmpeg) +- Mock StorageService (download/upload methods) +- Mock RedisService (progress reporting) +- Test `transcodeVideo()` method with various job configurations +- Test multi-resolution transcoding loop +- Test master playlist generation +- Test progress reporting at milestones +- Test error handling scenarios (download errors, FFmpeg errors, upload errors) +- Test temporary file cleanup in success and failure scenarios + +**Test File**: `apps/worker/src/app/transcoder/transcoder.service.spec.ts` + +**Test Cases**: +- Transcode video succeeds for single resolution +- Transcode video succeeds for multiple resolutions +- Master playlist is generated correctly +- Progress is reported at correct milestones +- Download errors are handled correctly +- FFmpeg errors are handled correctly +- Upload errors are handled correctly +- Temporary files are cleaned up on success +- Temporary files are cleaned up on failure +- Error categorization works correctly + +**Acceptance Criteria**: +- All test cases pass +- Code coverage is 70%+ for TranscoderService +- Tests use proper mocking (no real FFmpeg execution) +- Tests are fast (< 2 seconds total) + +--- + +#### FR1.5: Queue Consumer Service Unit Tests + +**Requirement**: Create comprehensive unit tests for QueueConsumerService. + +**Details**: +- Mock RabbitMQ channel and messages +- Mock TranscoderService +- Mock VideosService +- Test message consumption +- Test message parsing and validation +- Test retry logic (retry count tracking) +- Test error handling (retryable vs non-retryable errors) +- Test status updates (QUEUED β†’ PROCESSING β†’ COMPLETED/FAILED) +- Test graceful shutdown + +**Test File**: `apps/worker/src/app/queue-consumer/queue-consumer.service.spec.ts` + +**Test Cases**: +- Message consumption succeeds +- Message parsing succeeds for valid JSON +- Message parsing fails for invalid JSON (nack without requeue) +- Retry logic increments retry count correctly +- Retry logic stops after max retries +- Retryable errors are requeued +- Non-retryable errors are not requeued +- Status updates are called correctly +- Graceful shutdown finishes current job + +**Acceptance Criteria**: +- All test cases pass +- Code coverage is 70%+ for QueueConsumerService +- Tests use proper mocking (no real RabbitMQ connection) +- Tests are fast (< 1 second total) + +--- + +#### FR1.6: Enhanced Integration Tests + +**Requirement**: Enhance existing e2e tests with comprehensive end-to-end flows. + +**Details**: +- Update `apps/api-e2e/src/api/api.spec.ts` with real upload/download flow +- Test complete workflow: Upload β†’ Queue β†’ Transcode β†’ Progress β†’ Playlist +- Test error scenarios: Invalid files, connection failures, retry logic +- Test multi-resolution transcoding +- Test progress tracking accuracy + +**Test File**: `apps/api-e2e/src/api/api.spec.ts` + +**Test Cases**: +- Upload video succeeds and returns video ID +- Video appears in database with correct status +- Video file is stored in MinIO +- Job is queued in RabbitMQ +- Worker processes job successfully +- Progress updates are accurate +- All resolutions are generated +- Master playlist is accessible +- Progress endpoint returns correct data +- Invalid file upload is rejected +- Connection failures are handled gracefully + +**Acceptance Criteria**: +- All test cases pass +- Tests use real infrastructure (Docker containers) +- Tests clean up after themselves +- Tests are reliable (no flaky tests) + +--- + +### Phase 2: Production Readiness + +#### FR2.1: Dockerfile for API Application + +**Requirement**: Create multi-stage Dockerfile for API application. + +**Details**: +- Use Node.js base image +- Multi-stage build (build stage + production stage) +- Install dependencies +- Build application +- Copy only necessary files to production image +- Set proper working directory +- Expose port 3000 +- Set non-root user for security +- Add health check + +**File**: `apps/api/Dockerfile` + +**Dockerfile Structure**: +```dockerfile +# Build stage +FROM node:20-alpine AS builder +WORKDIR /app +COPY package*.json ./ +COPY nx.json tsconfig.base.json ./ +COPY apps/api ./apps/api +RUN npm ci +RUN npx nx build api + +# Production stage +FROM node:20-alpine +WORKDIR /app +COPY --from=builder /app/dist/apps/api ./dist/apps/api +COPY --from=builder /app/node_modules ./node_modules +COPY --from=builder /app/package.json ./ +EXPOSE 3000 +USER node +CMD ["node", "dist/apps/api/main.js"] +``` + +**Acceptance Criteria**: +- Dockerfile builds successfully +- Image size is optimized (< 500MB) +- Application runs correctly in container +- Health check works + +--- + +#### FR2.2: Dockerfile for Worker Application + +**Requirement**: Create Dockerfile for Worker application with FFmpeg. + +**Details**: +- Use Node.js base image +- Install FFmpeg system package +- Multi-stage build (build stage + production stage) +- Install dependencies +- Build application +- Copy only necessary files to production image +- Set proper working directory +- Set non-root user for security +- Verify FFmpeg is available + +**File**: `apps/worker/Dockerfile` + +**Dockerfile Structure**: +```dockerfile +# Build stage +FROM node:20-alpine AS builder +WORKDIR /app +COPY package*.json ./ +COPY nx.json tsconfig.base.json ./ +COPY apps/worker ./apps/worker +RUN npm ci +RUN npx nx build worker + +# Production stage +FROM node:20-alpine +RUN apk add --no-cache ffmpeg +WORKDIR /app +COPY --from=builder /app/dist/apps/worker ./dist/apps/worker +COPY --from=builder /app/node_modules ./node_modules +COPY --from=builder /app/package.json ./ +USER node +CMD ["node", "dist/apps/worker/main.js"] +``` + +**Acceptance Criteria**: +- Dockerfile builds successfully +- FFmpeg is available in container +- Image size is optimized (< 600MB) +- Application runs correctly in container + +--- + +#### FR2.3: Production Docker Compose + +**Requirement**: Create production Docker Compose configuration. + +**Details**: +- Use environment variables for all configuration +- Add health checks for all services +- Configure proper networking +- Add restart policies +- Configure resource limits +- Add logging configuration + +**File**: `docker-compose.prod.yml` + +**Acceptance Criteria**: +- All services start correctly +- Health checks pass +- Services can communicate with each other +- Configuration is environment-based + +--- + +#### FR2.4: Health Check Endpoint + +**Requirement**: Add health check endpoint to API. + +**Details**: +- Create `GET /health` endpoint +- Check database connection (PostgreSQL) +- Check Redis connection +- Check MinIO connection +- Check RabbitMQ connection +- Return 200 if all healthy +- Return 503 if any service unavailable +- Include service status in response + +**Endpoint**: `GET /health` + +**Response Format**: +```json +{ + "status": "ok" | "degraded" | "down", + "services": { + "database": { "status": "up" | "down", "latency": 5 }, + "redis": { "status": "up" | "down", "latency": 2 }, + "minio": { "status": "up" | "down", "latency": 10 }, + "rabbitmq": { "status": "up" | "down", "latency": 3 } + }, + "timestamp": "2024-01-01T12:00:00Z" +} +``` + +**Acceptance Criteria**: +- Endpoint returns correct status +- All service checks work correctly +- Response time is fast (< 100ms) +- Endpoint is documented in Swagger + +--- + +#### FR2.5: Environment Configuration + +**Requirement**: Create `.env.example` and validate environment variables. + +**Details**: +- Create `.env.example` with all required variables +- Document each variable with comments +- Add environment variable validation at startup +- Fail fast if required variables are missing +- Log configuration on startup (without sensitive values) + +**File**: `.env.example` + +**Variables to Document**: +- Database configuration (host, port, user, password, database) +- Redis configuration (host, port) +- MinIO configuration (endpoint, port, access key, secret key, buckets) +- RabbitMQ configuration (host, port, user, password, queue name) +- Application configuration (port, node env) +- FFmpeg configuration (path, temp directory) +- Transcode configuration (max retries, retry delay) + +**Acceptance Criteria**: +- `.env.example` includes all required variables +- Variables are documented with descriptions +- Validation fails fast on missing required variables +- Configuration is logged on startup + +--- + +### Phase 3: Security & Performance + +#### FR3.1: Rate Limiting + +**Requirement**: Add rate limiting to API endpoints. + +**Details**: +- Install `@nestjs/throttler` package +- Configure rate limiting module +- Set limits for upload endpoint (e.g., 10 requests per minute) +- Set limits for other endpoints (e.g., 100 requests per minute) +- Return 429 Too Many Requests when limit exceeded +- Include rate limit headers in response + +**Configuration**: +- Upload endpoint: 10 requests/minute +- Progress endpoint: 60 requests/minute +- Other endpoints: 100 requests/minute + +**Acceptance Criteria**: +- Rate limiting works correctly +- Limits are configurable +- Rate limit headers are included in responses +- 429 status is returned when limit exceeded + +--- + +#### FR3.2: Input Validation + +**Requirement**: Enhance DTOs with validation decorators. + +**Details**: +- Install `class-validator` and `class-transformer` packages +- Add validation to `UploadVideoDto` +- Validate file type (video/* mime types) +- Validate file size (max 5GB) +- Validate video ID format (UUID) +- Return 400 Bad Request for invalid input + +**File**: `apps/api/src/app/videos/dto/upload-video.dto.ts` + +**Validation Rules**: +- File is required +- File type must be video/* +- File size must be <= 5GB +- Video ID must be valid UUID format + +**Acceptance Criteria**: +- Invalid uploads are rejected +- Error messages are clear +- Validation works for all endpoints +- Validation is documented in Swagger + +--- + +#### FR3.3: CORS Configuration + +**Requirement**: Make CORS configuration environment-based. + +**Details**: +- Move CORS configuration from hardcoded `*` to environment variables +- Support multiple allowed origins +- Configure allowed methods and headers +- Support credentials if needed + +**Configuration**: +- `CORS_ORIGIN` environment variable (comma-separated origins) +- Default to `*` in development +- Require explicit origins in production + +**Acceptance Criteria**: +- CORS works correctly +- Configuration is environment-based +- Multiple origins are supported +- Headers are properly configured + +--- + +### Phase 4: Stretch Goals + +#### FR4.1: Job Cancellation + +**Requirement**: Add job cancellation endpoint. + +**Details**: +- Create `POST /videos/:id/cancel` endpoint +- Mark job as cancelled in Redis +- Worker checks cancellation flag before processing +- Cleanup in-progress transcoding jobs +- Update video status to CANCELLED + +**Endpoint**: `POST /videos/:id/cancel` + +**Response Format**: +```json +{ + "success": true, + "message": "Job cancellation requested", + "videoId": "123e4567-e89b-12d3-a456-426614174000" +} +``` + +**Acceptance Criteria**: +- Endpoint works correctly +- Cancellation is checked by worker +- In-progress jobs are cleaned up +- Status is updated correctly + +--- + +#### FR4.2: Video Metadata Extraction + +**Requirement**: Extract and store video metadata. + +**Details**: +- Use FFprobe to extract metadata (duration, resolution, codec, bitrate) +- Store metadata in Video entity +- Add `GET /videos/:id/metadata` endpoint +- Display metadata in dashboard + +**Metadata Fields**: +- Duration (seconds) +- Resolution (width x height) +- Codec (video codec, audio codec) +- Bitrate (video bitrate, audio bitrate) +- Frame rate +- File size + +**Acceptance Criteria**: +- Metadata is extracted correctly +- Metadata is stored in database +- Endpoint returns metadata +- Dashboard displays metadata + +--- + +#### FR4.3: Enhanced API Endpoints + +**Requirement**: Add missing API endpoints. + +**Details**: +- `GET /videos/:id` - Get video details with status +- `GET /videos` - List all videos with pagination +- `DELETE /videos/:id` - Delete video and cleanup storage + +**Endpoints**: +- `GET /videos/:id` - Returns video entity with all fields +- `GET /videos?page=1&limit=10` - Returns paginated list of videos +- `DELETE /videos/:id` - Deletes video record and MinIO objects + +**Acceptance Criteria**: +- All endpoints work correctly +- Pagination works +- Deletion cleans up storage +- Endpoints are documented in Swagger + +--- + +### Phase 5: Documentation & Polish + +#### FR5.1: Enhanced API Documentation + +**Requirement**: Enhance Swagger documentation. + +**Details**: +- Add examples to all endpoints +- Document all response schemas +- Document error responses +- Add request/response examples +- Group endpoints logically + +**Acceptance Criteria**: +- All endpoints have examples +- Error responses are documented +- Documentation is clear and complete +- Examples are realistic + +--- + +#### FR5.2: README Enhancement + +**Requirement**: Enhance README with comprehensive documentation. + +**Details**: +- Add architecture diagram (Mermaid) +- Add deployment guide +- Add troubleshooting section +- Add development setup guide +- Add API usage examples + +**Sections to Add**: +- Architecture Overview +- Deployment Guide +- Development Setup +- API Usage Examples +- Troubleshooting +- Contributing + +**Acceptance Criteria**: +- README is comprehensive +- Architecture diagram is clear +- Deployment guide is complete +- Troubleshooting section is helpful + +--- + +## Implementation Details + +### Testing Strategy + +**Unit Tests**: +- Use Jest as testing framework +- Mock external dependencies (MinIO, RabbitMQ, Redis, FFmpeg) +- Aim for 70%+ code coverage +- Focus on error paths and edge cases +- Keep tests fast (< 1 second per test file) + +**Integration Tests**: +- Use real Docker containers for infrastructure +- Test end-to-end flows +- Clean up after tests +- Make tests reliable (no flaky tests) + +**Test Organization**: +- Unit tests: `*.spec.ts` files next to source files +- Integration tests: `apps/*-e2e/src/**/*.spec.ts` +- Test utilities: Shared mocks and helpers + +--- + +### Docker Strategy + +**Multi-Stage Builds**: +- Build stage: Install dependencies and build +- Production stage: Copy only necessary files +- Minimize image size +- Use Alpine Linux for smaller images + +**Image Optimization**: +- Use `.dockerignore` to exclude unnecessary files +- Use layer caching effectively +- Minimize number of layers +- Use specific version tags + +--- + +### Security Considerations + +**Rate Limiting**: +- Prevent abuse of upload endpoint +- Protect against DDoS attacks +- Configurable limits per endpoint + +**Input Validation**: +- Validate all user input +- Reject invalid files early +- Prevent malicious uploads + +**CORS**: +- Restrict origins in production +- Don't use `*` in production +- Configure allowed methods and headers + +--- + +## Testing Requirements + +### Unit Testing + +**StorageService Tests**: +- Mock MinIO client +- Test all methods +- Test error scenarios +- Achieve 80%+ coverage + +**QueueService Tests**: +- Mock amqplib +- Test job publishing +- Test error handling +- Achieve 80%+ coverage + +**RedisService Tests**: +- Mock ioredis +- Test progress tracking +- Test error handling +- Achieve 80%+ coverage + +**TranscoderService Tests**: +- Mock FFmpeg and storage +- Test transcoding flow +- Test error handling +- Achieve 70%+ coverage + +**QueueConsumerService Tests**: +- Mock RabbitMQ and services +- Test message handling +- Test retry logic +- Achieve 70%+ coverage + +### Integration Testing + +**End-to-End Flow**: +- Upload video β†’ Verify queued +- Poll progress β†’ Verify updates +- Check playlist β†’ Verify accessible +- Test error scenarios + +**Error Scenarios**: +- Invalid file upload +- Connection failures +- Retry logic +- Job cancellation + +### Manual Testing + +**Test Cases**: +1. Upload small video (10MB) β†’ Verify transcoding +2. Upload large video (500MB) β†’ Verify transcoding +3. Test progress tracking accuracy +4. Test all resolutions are generated +5. Test master playlist +6. Test error handling +7. Test job cancellation +8. Test metadata extraction + +--- + +## Success Criteria + +### Functional Success + +- βœ… Unit tests exist for all core services +- βœ… Integration tests cover end-to-end flows +- βœ… Test coverage is 70%+ on critical services +- βœ… Dockerfiles build and run correctly +- βœ… Health check endpoint works +- βœ… Rate limiting works +- βœ… Input validation works +- βœ… Job cancellation works +- βœ… Metadata extraction works +- βœ… All endpoints are documented + +### Non-Functional Success + +- βœ… Tests are fast (< 1 second per unit test file) +- βœ… Docker images are optimized (< 600MB) +- βœ… Health check is fast (< 100ms) +- βœ… Rate limiting is configurable +- βœ… Documentation is complete +- βœ… Code is maintainable + +--- + +## Dependencies + +### External Dependencies + +- **@nestjs/throttler**: Rate limiting +- **class-validator**: Input validation +- **class-transformer**: DTO transformation +- **@nestjs/terminus**: Health checks (optional) +- **jest**: Testing framework +- **@types/jest**: Jest TypeScript types + +### Internal Dependencies + +- All existing services and modules +- Docker infrastructure +- Configuration system + +--- + +## Risks & Mitigations + +### Risk 1: Test Coverage Goals Too Ambitious + +**Mitigation**: Start with critical paths, add more tests incrementally + +### Risk 2: Docker Image Size Too Large + +**Mitigation**: Use multi-stage builds, Alpine Linux, optimize layers + +### Risk 3: Health Checks Too Slow + +**Mitigation**: Use timeouts, cache results, make checks lightweight + +### Risk 4: Rate Limiting Too Restrictive + +**Mitigation**: Make limits configurable, start with generous limits + +--- + +## Timeline & Effort Estimate + +### Phase 1: Testing Infrastructure (Week 1) +- **Effort**: 12-16 hours +- **Tasks**: Unit tests for all services, integration tests + +### Phase 2: Production Readiness (Week 2) +- **Effort**: 10-12 hours +- **Tasks**: Dockerfiles, health checks, environment configuration + +### Phase 3: Security & Performance (Week 3) +- **Effort**: 8-10 hours +- **Tasks**: Rate limiting, validation, CORS + +### Phase 4: Stretch Goals (Week 4) +- **Effort**: 8-10 hours +- **Tasks**: Job cancellation, metadata extraction, enhanced endpoints + +### Phase 5: Documentation & Polish (Week 4) +- **Effort**: 4-6 hours +- **Tasks**: API docs, README, code documentation + +**Total Effort**: 42-54 hours (approximately 4 weeks at 10-15 hours/week) + +--- + +## Conclusion + +This PRD defines the requirements for completing the StreamForge project, focusing on testing, production readiness, security, and stretch goals. The implementation is organized into 5 phases, with clear success criteria and acceptance criteria for each requirement. The project is functionally complete, and this completion phase will make it production-ready and maintainable. + +**Next Steps**: +1. Review and approve this PRD +2. Prioritize phases based on business needs +3. Begin implementation with Phase 1 (Testing Infrastructure) +4. Iterate and refine based on feedback + +--- + +**Document Version**: 1.0 +**Last Updated**: 2024 +**Status**: Active + diff --git a/docs/done/PRD-MULTI-RESOLUTION-ERROR-HANDLING.md b/docs/done/PRD-MULTI-RESOLUTION-ERROR-HANDLING.md new file mode 100644 index 0000000..1cd1e1b --- /dev/null +++ b/docs/done/PRD-MULTI-RESOLUTION-ERROR-HANDLING.md @@ -0,0 +1,1070 @@ +# Product Requirements Document: Multi-Resolution Transcoding & Error Handling + +## Executive Summary + +This document defines the requirements for implementing multi-resolution transcoding and enhanced error handling for the StreamForge video transcoding platform. The system will transcode videos to three resolutions (360p, 720p, 1080p), generate a master HLS playlist that enables adaptive bitrate streaming, and implement robust error handling with cleanup and retry logic. This feature is critical for providing optimal video quality across different network conditions and ensuring system reliability during transcoding operations. + +At a high level, the implementation will be completed in the following steps: + +1. **Update TranscodeJobData structure** + Extend the job data structure to include a resolutions array, replacing the single profile field with multiple resolution configurations. + +2. **Implement multi-resolution transcoding loop** + Update TranscoderService to loop through 360p, 720p, and 1080p resolutions, transcoding each sequentially and uploading all segments. + +3. **Generate master playlist (index.m3u8)** + Create a master HLS playlist that references all three resolution playlists with appropriate bandwidth tags for adaptive streaming. + +4. **Enhance error handling in queue consumer** + Add comprehensive try/catch blocks, ensure status updates to FAILED on errors, and implement cleanup in finally blocks. + +5. **Implement basic retry logic** + Add retry count tracking and re-queue failed jobs once before marking as permanently failed. + +6. **Update progress reporting for multi-resolution** + Adjust progress milestones to account for multiple transcoding operations. + +7. **Test end-to-end** + Verify all three resolutions are generated, master playlist is correct, and error handling works as expected. + +**Timeline**: Day 14 (2-3 hours) +**Effort Estimate**: 2-3 hours +**Priority**: High (Enables adaptive streaming and improves system reliability) + +--- + +## Background & Context + +### Why Multi-Resolution Transcoding is Needed + +StreamForge currently transcodes videos to a single resolution (720p), which limits the viewing experience for users with varying network conditions. Multi-resolution transcoding enables: + +- **Adaptive Bitrate Streaming (ABR)**: Video players can automatically switch between quality levels based on available bandwidth +- **Better User Experience**: Users on slow connections get 360p, while users on fast connections get 1080p +- **Reduced Buffering**: Lower quality streams load faster, reducing playback interruptions +- **Industry Standard**: HLS with multiple resolutions is the standard for modern video streaming platforms + +### Why Enhanced Error Handling is Needed + +The current error handling in the queue consumer has basic retry logic but lacks: +- Comprehensive cleanup of temporary files on all error paths +- Clear error categorization for different failure types +- Proper status updates in all error scenarios +- Retry count tracking to prevent infinite retries + +Enhanced error handling ensures: +- **System Reliability**: Failed jobs don't leave orphaned files or stuck states +- **Resource Cleanup**: Temporary files are always cleaned up, preventing disk space issues +- **Clear Error Reporting**: Users and operators can understand why jobs failed +- **Controlled Retries**: Jobs are retried once before being marked as permanently failed + +### Integration with Existing System + +The multi-resolution transcoding integrates with: +- **Transcoder Service** (`apps/worker/src/app/transcoder/transcoder.service.ts`): Currently handles single 720p transcoding, will be extended for multiple resolutions +- **Queue Consumer Service** (`apps/worker/src/app/queue-consumer/queue-consumer.service.ts`): Handles job consumption and status updates, will be enhanced with better error handling +- **Queue Service** (`apps/api/src/app/queue/queue.service.ts`): Publishes jobs, will be updated to include resolutions array +- **Storage Service** (`apps/worker/src/app/storage/storage.service.ts`): Uploads HLS files, will upload multiple resolution playlists and segments +- **Redis Service** (`apps/worker/src/app/redis/redis.service.ts`): Tracks progress, will be updated for multi-resolution progress reporting +- **Video Entity** (`apps/api/src/app/videos/entities/video.entity.ts`): Tracks video status, status updates will be enhanced +- **API Controller** (`apps/api/src/app/app.controller.ts`): Serves playlists, will serve master playlist instead of single resolution + +### Architecture Context + +The multi-resolution transcoding follows the StreamForge architecture principle: **Multi-Resolution: Transcode to multiple resolutions, generate master playlist for adaptive streaming**. This enables: +- Sequential transcoding of each resolution (simpler than parallel, avoids resource contention) +- Single master playlist that references all resolutions +- Progress tracking across all resolutions +- Cleanup of all temporary files regardless of success/failure + +### Current State + +- Transcoder service handles single 720p resolution +- FFmpeg transcoding is working with HLS output +- Queue consumer has basic error handling with retry logic +- Temporary file cleanup exists in finally block +- Progress reporting tracks overall job progress +- Single resolution playlist is served via API + +### Missing Components + +- Multi-resolution loop in transcoder service +- Master playlist generation +- Resolution configuration structure +- Enhanced error handling with comprehensive cleanup +- Retry count tracking in job messages +- Progress milestones adjusted for multi-resolution workflow + +--- + +## Objectives & Goals + +### Primary Objectives + +1. **Multi-Resolution Transcoding**: Transcode videos to 360p, 720p, and 1080p resolutions +2. **Master Playlist Generation**: Create `index.m3u8` that references all resolutions with bandwidth tags +3. **Enhanced Error Handling**: Implement comprehensive try/catch blocks with proper cleanup +4. **Retry Logic**: Re-queue failed jobs once before marking as permanently failed +5. **Progress Tracking**: Update progress reporting to account for multiple transcoding operations +6. **Resource Cleanup**: Ensure all temporary files are cleaned up in all error scenarios + +### Success Criteria + +- βœ… Videos are transcoded to all three resolutions (360p, 720p, 1080p) +- βœ… All resolution playlists and segments are uploaded to MinIO +- βœ… Master playlist (`index.m3u8`) is generated and uploaded +- βœ… Master playlist includes bandwidth tags for adaptive streaming +- βœ… Error handling catches all error types and updates status appropriately +- βœ… Temporary files are cleaned up in all scenarios (success and failure) +- βœ… Failed jobs are retried once before being marked as permanently failed +- βœ… Progress reporting accounts for multiple resolutions +- βœ… Test: Upload video β†’ Check all 3 resolutions in MinIO β†’ Verify master playlist + +### Key Deliverables + +- Updated `apps/worker/src/app/transcoder/dto/transcode-job-data.dto.ts` - Resolution array structure +- Updated `apps/worker/src/app/transcoder/transcoder.service.ts` - Multi-resolution loop and master playlist generation +- Updated `apps/api/src/app/queue/queue.service.ts` - Include resolutions in job message +- Updated `apps/worker/src/app/queue-consumer/queue-consumer.service.ts` - Enhanced error handling +- Updated `apps/worker/src/main.ts` - Error handling improvements (if needed) +- Master playlist generation utility +- Working multi-resolution transcoding with all three resolutions +- Test verification: Upload video β†’ Check all resolutions β†’ Verify master playlist + +--- + +## Functional Requirements + +### FR1: Resolution Configuration Structure + +**Requirement**: The job data structure must support multiple resolutions instead of a single profile. + +**Details**: +- Update `TranscodeJobData` interface to include `resolutions` array +- Each resolution should have: `height` (number), `width` (calculated), `bandwidth` (estimated) +- Default resolutions: `[{360p}, {720p}, {1080p}]` +- Maintain backward compatibility during transition (support both `profile` and `resolutions`) + +**Resolution Configuration**: +```typescript +interface ResolutionConfig { + height: number; // 360, 720, 1080 + width: number; // Calculated based on aspect ratio + bandwidth: number; // Estimated bandwidth in bits per second + label: string; // '360p', '720p', '1080p' +} + +interface TranscodeJobData { + videoId: string; + sourceBucket: string; + sourceKey: string; + targetBucket: string; + resolutions: ResolutionConfig[]; // New field + profile?: 'HLS_720P'; // Deprecated, kept for compatibility +} +``` + +**Resolution Specifications**: +- **360p**: Height 360, Width 640 (16:9), Bandwidth ~800000 (800 kbps) +- **720p**: Height 720, Width 1280 (16:9), Bandwidth ~2500000 (2.5 Mbps) +- **1080p**: Height 1080, Width 1920 (16:9), Bandwidth ~5000000 (5 Mbps) + +### FR2: Multi-Resolution Transcoding Loop + +**Requirement**: Transcoder service must loop through all resolutions and transcode each one. + +**Details**: +- Loop through `resolutions` array in order: 360p β†’ 720p β†’ 1080p +- For each resolution: + - Create output directory: `{tempDir}/output/{resolution}` + - Execute FFmpeg with resolution-specific parameters + - Upload all segments and playlist to MinIO: `{targetBucket}/{videoId}/{resolution}/` +- Continue to next resolution even if one fails (log error, mark resolution as failed) +- Track which resolutions succeeded/failed for error reporting + +**Transcoding Flow**: +``` +For each resolution in [360p, 720p, 1080p]: + 1. Create output directory + 2. Execute FFmpeg transcoding + 3. Upload playlist and segments to MinIO + 4. Report progress for this resolution + 5. Continue to next resolution +``` + +**FFmpeg Parameters per Resolution**: +- **360p**: `.size('640x360')` with bandwidth-optimized settings +- **720p**: `.size('1280x720')` (current implementation) +- **1080p**: `.size('1920x1080')` with high-quality settings + +**Error Handling**: +- If one resolution fails, log error and continue with next resolution +- Track failed resolutions in error message +- Mark job as failed only if all resolutions fail + +### FR3: Master Playlist Generation + +**Requirement**: System must generate a master HLS playlist (`index.m3u8`) that references all resolution playlists. + +**Details**: +- Generate `index.m3u8` file after all resolutions are transcoded +- Include all successfully transcoded resolutions +- Add bandwidth tags (`#EXT-X-STREAM-INF`) for each resolution +- Upload master playlist to MinIO: `{targetBucket}/{videoId}/index.m3u8` +- Master playlist format must conform to HLS specification + +**Master Playlist Format**: +``` +#EXTM3U +#EXT-X-VERSION:3 +#EXT-X-STREAM-INF:BANDWIDTH=800000,RESOLUTION=640x360 +360p/360p.m3u8 +#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=1280x720 +720p/720p.m3u8 +#EXT-X-STREAM-INF:BANDWIDTH=5000000,RESOLUTION=1920x1080 +1080p/1080p.m3u8 +``` + +**Bandwidth Tags**: +- `BANDWIDTH`: Estimated bandwidth in bits per second +- `RESOLUTION`: Video resolution (widthxheight) +- Order: Lowest to highest bandwidth (360p β†’ 720p β†’ 1080p) + +**Playlist Paths**: +- Master playlist: `{videoId}/index.m3u8` +- Resolution playlists: `{videoId}/360p/360p.m3u8`, `{videoId}/720p/720p.m3u8`, `{videoId}/1080p/1080p.m3u8` +- Relative paths in master playlist (e.g., `360p/360p.m3u8`) + +### FR4: Enhanced Error Handling + +**Requirement**: Queue consumer must have comprehensive error handling with proper cleanup and status updates. + +**Details**: +- Wrap entire job processing in try/catch block +- Update status to `FAILED` in database on any error +- Cleanup temporary files in `finally` block (always runs) +- Log detailed error information with context +- Categorize errors (download, transcode, upload, internal) +- Attach error information to error object for retry logic + +**Error Handling Flow**: +``` +try { + // Process job + await transcoderService.transcodeVideo(job); + // Update status to COMPLETED +} catch (error) { + // Categorize error + // Update status to FAILED + // Log error details + // Determine if retryable +} finally { + // Cleanup temp files (always runs) + // Remove from in-flight jobs +} +``` + +**Error Categories**: +- `DOWNLOAD_ERROR`: Failed to download source video +- `TRANSCODE_ERROR`: FFmpeg transcoding failed +- `UPLOAD_ERROR`: Failed to upload to MinIO +- `INTERNAL_ERROR`: Unexpected system error +- `VALIDATION_ERROR`: Invalid job data (non-retryable) + +**Status Updates**: +- On error: Update database status to `FAILED` +- Include error message in status update +- Update Redis status to `FAILED` with error message +- Log error with full stack trace + +### FR5: Temporary File Cleanup + +**Requirement**: All temporary files must be cleaned up in all scenarios (success and failure). + +**Details**: +- Cleanup must happen in `finally` block (always executes) +- Remove entire temp directory: `{TEMP_BASE}/{videoId}/{jobId}/` +- Handle cleanup errors gracefully (log but don't throw) +- Cleanup should be idempotent (safe to call multiple times) +- Use existing `removeTempDir()` utility function + +**Cleanup Flow**: +``` +finally { + if (tempDir) { + try { + await removeTempDir(tempDir); + logger.log(`Temp directory cleaned up: ${tempDir}`); + } catch (cleanupError) { + logger.error(`Failed to cleanup temp directory: ${cleanupError.message}`); + // Don't throw - cleanup failure shouldn't fail the job + } + } +} +``` + +**Cleanup Scope**: +- Input video file +- All output directories (`output/360p/`, `output/720p/`, `output/1080p/`) +- All HLS playlists and segments +- Master playlist temporary file +- Entire job temp directory + +### FR6: Basic Retry Logic + +**Requirement**: Failed jobs must be retried once before being marked as permanently failed. + +**Details**: +- Track retry count in job message headers: `x-retry-count` +- Increment retry count on failure +- Re-queue job if retry count < 1 (retry once) +- Mark as permanently failed if retry count >= 1 +- Non-retryable errors (validation errors) should not be retried + +**Retry Logic Flow**: +``` +On error: + 1. Get current retry count from message headers (default: 0) + 2. Check if error is retryable + 3. If retryable and retry count < 1: + - Increment retry count + - Update message headers + - Nack with requeue (true) + 4. If not retryable or retry count >= 1: + - Update status to FAILED + - Nack without requeue (false) +``` + +**Retry Count Tracking**: +- Store in RabbitMQ message headers: `x-retry-count: number` +- Default value: 0 (first attempt) +- Increment on each retry +- Max retries: 1 (retry once) + +**Non-Retryable Errors**: +- Validation errors (invalid job data) +- Malformed messages +- Missing required fields +- Invalid video ID + +### FR7: Progress Reporting for Multi-Resolution + +**Requirement**: Progress reporting must account for multiple transcoding operations. + +**Details**: +- Adjust progress milestones for multi-resolution workflow +- Report progress for each resolution completion +- Overall progress: Download (0-10%), Transcode (10-90%), Upload (90-100%) +- Within transcode phase: Each resolution contributes equally (10-30%, 30-50%, 50-70%) +- Master playlist generation: 90-95% +- Final upload: 95-100% + +**Progress Milestones**: +``` +0% β†’ Download starts +10% β†’ Download complete +10% β†’ 360p transcoding starts +20% β†’ 360p transcoding complete +20% β†’ 720p transcoding starts +40% β†’ 720p transcoding complete +40% β†’ 1080p transcoding starts +70% β†’ 1080p transcoding complete +70% β†’ Upload 360p starts +80% β†’ Upload 360p complete +80% β†’ Upload 720p starts +85% β†’ Upload 720p complete +85% β†’ Upload 1080p starts +90% β†’ Upload 1080p complete +90% β†’ Master playlist generation +95% β†’ Master playlist upload +100% β†’ Job complete +``` + +**Progress Calculation**: +- Download: 0-10% (10% of total) +- Transcode: 10-70% (60% of total, 20% per resolution) +- Upload: 70-95% (25% of total, ~8% per resolution) +- Master playlist: 95-100% (5% of total) + +--- + +## Technical Specifications + +### Technology Stack + +- **FFmpeg**: Video transcoding (already integrated) +- **fluent-ffmpeg**: Node.js FFmpeg wrapper (already integrated) +- **HLS**: HTTP Live Streaming protocol (already supported) +- **Node.js**: File system operations for playlist generation +- **NestJS**: Service and module patterns +- **TypeScript**: Full type safety + +### Resolution Configuration + +**Resolution Definitions**: +```typescript +const RESOLUTIONS: ResolutionConfig[] = [ + { + height: 360, + width: 640, + bandwidth: 800000, // 800 kbps + label: '360p', + }, + { + height: 720, + width: 1280, + bandwidth: 2500000, // 2.5 Mbps + label: '720p', + }, + { + height: 1080, + width: 1920, + bandwidth: 5000000, // 5 Mbps + label: '1080p', + }, +]; +``` + +**FFmpeg Parameters per Resolution**: +- **360p**: `.size('640x360')`, lower bitrate, faster encoding +- **720p**: `.size('1280x720')`, medium bitrate (current settings) +- **1080p**: `.size('1920x1080')`, higher bitrate, slower encoding + +### Master Playlist Format + +**HLS Master Playlist Specification**: +- File extension: `.m3u8` +- MIME type: `application/vnd.apple.mpegurl` +- Must start with `#EXTM3U` +- Each stream variant: `#EXT-X-STREAM-INF` tag followed by playlist path +- Bandwidth in bits per second +- Resolution in `WIDTHxHEIGHT` format + +**Master Playlist Generation**: +```typescript +function generateMasterPlaylist(resolutions: ResolutionConfig[]): string { + let playlist = '#EXTM3U\n#EXT-X-VERSION:3\n'; + + for (const resolution of resolutions) { + playlist += `#EXT-X-STREAM-INF:BANDWIDTH=${resolution.bandwidth},RESOLUTION=${resolution.width}x${resolution.height}\n`; + playlist += `${resolution.label}/${resolution.label}.m3u8\n`; + } + + return playlist; +} +``` + +### Error Handling Structure + +**Error Categorization**: +```typescript +enum ErrorCategory { + DOWNLOAD_ERROR = 'DOWNLOAD_ERROR', + TRANSCODE_ERROR = 'TRANSCODE_ERROR', + UPLOAD_ERROR = 'UPLOAD_ERROR', + INTERNAL_ERROR = 'INTERNAL_ERROR', + VALIDATION_ERROR = 'VALIDATION_ERROR', +} + +interface ErrorInfo { + category: ErrorCategory; + retryable: boolean; + phase: string; + message: string; +} +``` + +**Retry Decision Logic**: +- `DOWNLOAD_ERROR`: Retryable (network issues, temporary MinIO unavailability) +- `TRANSCODE_ERROR`: Retryable (FFmpeg crashes, codec issues) +- `UPLOAD_ERROR`: Retryable (network issues, MinIO unavailability) +- `INTERNAL_ERROR`: Retryable (unexpected errors, may be transient) +- `VALIDATION_ERROR`: Non-retryable (invalid data, won't succeed on retry) + +### File Structure + +**MinIO Structure After Transcoding**: +``` +processed-videos/ + {videoId}/ + index.m3u8 # Master playlist + 360p/ + 360p.m3u8 # 360p playlist + segment-00001.ts # 360p segments + segment-00002.ts + ... + 720p/ + 720p.m3u8 # 720p playlist + segment-00001.ts # 720p segments + segment-00002.ts + ... + 1080p/ + 1080p.m3u8 # 1080p playlist + segment-00001.ts # 1080p segments + segment-00002.ts + ... +``` + +**Temporary Directory Structure**: +``` +{TEMP_BASE}/ + {videoId}/ + {jobId}/ + input/ + source.mp4 # Downloaded source video + output/ + 360p/ + 360p.m3u8 # 360p playlist + segment-00001.ts # 360p segments + ... + 720p/ + 720p.m3u8 # 720p playlist + segment-00001.ts # 720p segments + ... + 1080p/ + 1080p.m3u8 # 1080p playlist + segment-00001.ts # 1080p segments + ... + index.m3u8 # Master playlist (before upload) +``` + +--- + +## Architecture & Design + +### Data Flow + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ API β”‚ +β”‚ QueueServiceβ”‚ +β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”˜ + β”‚ Publish job with resolutions array + β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ RabbitMQ β”‚ +β”‚ Queue β”‚ +β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ Consume job + β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ QueueConsumerService β”‚ +β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ transcodeVideo(job) + β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ TranscoderService β”‚ +β”‚ β”‚ +β”‚ For each resolution:β”‚ +β”‚ 1. Download source β”‚ +β”‚ 2. Transcode β”‚ +β”‚ 3. Upload β”‚ +β”‚ β”‚ +β”‚ Generate master β”‚ +β”‚ playlist β”‚ +β”‚ β”‚ +β”‚ Upload master β”‚ +β”‚ playlist β”‚ +β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ + β”œβ”€β†’ FFmpeg (360p) + β”œβ”€β†’ FFmpeg (720p) + β”œβ”€β†’ FFmpeg (1080p) + └─→ Master playlist +``` + +### Multi-Resolution Transcoding Flow + +``` +transcodeVideo(job) + β”‚ + β”œβ”€β†’ Download source video (0-10%) + β”‚ + β”œβ”€β†’ For resolution in [360p, 720p, 1080p]: + β”‚ β”‚ + β”‚ β”œβ”€β†’ Create output directory + β”‚ β”œβ”€β†’ Execute FFmpeg transcoding + β”‚ β”œβ”€β†’ Upload playlist and segments + β”‚ └─→ Report progress + β”‚ + β”œβ”€β†’ Generate master playlist (90-95%) + β”œβ”€β†’ Upload master playlist (95-100%) + └─→ Cleanup temp files (finally) +``` + +### Error Handling Flow + +``` +QueueConsumerService.handleMessage() + β”‚ + β”œβ”€β†’ try { + β”‚ β”‚ + β”‚ β”œβ”€β†’ Parse and validate job + β”‚ β”œβ”€β†’ Update status to PROCESSING + β”‚ β”œβ”€β†’ Call transcoderService.transcodeVideo() + β”‚ β”œβ”€β†’ Update status to COMPLETED + β”‚ └─→ ACK message + β”‚ + └─→ catch (error) { + β”‚ + β”œβ”€β†’ Categorize error + β”œβ”€β†’ Get retry count + β”œβ”€β†’ If retryable and retry count < 1: + β”‚ └─→ Increment retry count, NACK with requeue + └─→ Else: + └─→ Update status to FAILED, NACK without requeue + } + β”‚ + └─→ finally { + └─→ Cleanup temp files + └─→ Remove from in-flight jobs + } +``` + +### Module Updates + +**TranscoderService Updates**: +- Add `transcodeMultipleResolutions()` method +- Update `transcodeVideo()` to call multi-resolution method +- Add `generateMasterPlaylist()` method +- Update progress reporting for multi-resolution workflow + +**QueueConsumerService Updates**: +- Enhance error handling in `handleMessage()` +- Add retry count tracking +- Ensure cleanup in finally block +- Improve error categorization + +**QueueService Updates**: +- Update `publishTranscodeJob()` to include resolutions array +- Maintain backward compatibility with profile field + +--- + +## Implementation Plan + +### Phase 1: Update Job Data Structure (30 minutes) + +1. **Update TranscodeJobData DTO** + - Add `resolutions: ResolutionConfig[]` field + - Keep `profile` field for backward compatibility (deprecated) + - Define `ResolutionConfig` interface + - Update type definitions + +2. **Update QueueService** + - Modify `publishTranscodeJob()` to include resolutions array + - Default to `[360p, 720p, 1080p]` if not specified + - Maintain backward compatibility + +3. **Test Job Publishing** + - Verify job messages include resolutions array + - Verify backward compatibility with profile field + +### Phase 2: Multi-Resolution Transcoding Loop (1 hour) + +1. **Update TranscoderService** + - Add `transcodeMultipleResolutions()` method + - Loop through resolutions array + - For each resolution: + - Create output directory + - Execute FFmpeg with resolution-specific parameters + - Upload playlist and segments + - Handle errors per resolution (continue on failure) + +2. **Update FFmpeg Execution** + - Extract FFmpeg command to helper method + - Add resolution parameter to FFmpeg command + - Update size and bitrate per resolution + +3. **Update Progress Reporting** + - Adjust progress milestones for multi-resolution + - Report progress after each resolution + - Update Redis progress at key milestones + +4. **Test Multi-Resolution Transcoding** + - Upload video and verify all three resolutions are created + - Verify all segments are uploaded to MinIO + - Check progress reporting + +### Phase 3: Master Playlist Generation (30 minutes) + +1. **Create Master Playlist Generator** + - Add `generateMasterPlaylist()` method + - Format HLS master playlist with bandwidth tags + - Include all successfully transcoded resolutions + +2. **Upload Master Playlist** + - Generate master playlist after all resolutions complete + - Upload to MinIO: `{videoId}/index.m3u8` + - Update progress to 100% + +3. **Test Master Playlist** + - Verify master playlist is generated correctly + - Verify master playlist is uploaded to MinIO + - Test master playlist in video player + +### Phase 4: Enhanced Error Handling (30 minutes) + +1. **Update QueueConsumerService** + - Enhance try/catch in `handleMessage()` + - Ensure cleanup in finally block + - Improve error categorization + - Add detailed error logging + +2. **Update TranscoderService** + - Ensure cleanup in finally block (already exists, verify) + - Improve error messages with context + - Track which resolutions failed + +3. **Test Error Handling** + - Test with invalid video (should fail gracefully) + - Test with FFmpeg error (should cleanup and update status) + - Test with upload error (should cleanup and update status) + +### Phase 5: Retry Logic (30 minutes) + +1. **Add Retry Count Tracking** + - Read retry count from message headers + - Increment retry count on failure + - Update message headers with new retry count + +2. **Update Retry Decision Logic** + - Check if error is retryable + - Check retry count (< 1 for retry) + - NACK with requeue if retryable and count < 1 + - NACK without requeue if not retryable or count >= 1 + +3. **Test Retry Logic** + - Test retryable error (should retry once) + - Test non-retryable error (should not retry) + - Test max retries (should mark as failed after 1 retry) + +### Phase 6: Testing & Verification (30 minutes) + +1. **End-to-End Testing** + - Upload video β†’ Verify all 3 resolutions in MinIO + - Verify master playlist is correct + - Test master playlist in video player + - Verify adaptive streaming works + +2. **Error Scenario Testing** + - Test with invalid video (should fail gracefully) + - Test with FFmpeg error (should cleanup and retry) + - Test with upload error (should cleanup and retry) + - Test max retries (should mark as failed) + +3. **Performance Testing** + - Verify multi-resolution transcoding doesn't significantly slow down jobs + - Check disk space usage (temp files should be cleaned up) + - Monitor memory usage during transcoding + +--- + +## Testing Requirements + +### Unit Tests + +**TranscoderService Tests**: +- Test multi-resolution loop processes all resolutions +- Test master playlist generation with all resolutions +- Test master playlist generation with partial failures +- Test progress reporting for multi-resolution +- Test error handling per resolution (continues on failure) + +**QueueConsumerService Tests**: +- Test error handling updates status to FAILED +- Test cleanup in finally block always runs +- Test retry logic increments retry count +- Test retry logic respects max retries +- Test non-retryable errors are not retried + +**Master Playlist Generator Tests**: +- Test master playlist format is correct +- Test bandwidth tags are included +- Test resolution tags are included +- Test playlist paths are relative + +### Integration Tests + +**End-to-End Flow**: +- Upload video β†’ Verify all 3 resolutions are created +- Verify master playlist references all resolutions +- Verify adaptive streaming works in video player +- Verify progress reporting accounts for all resolutions + +**Error Scenarios**: +- Invalid video β†’ Job fails, status updated, temp files cleaned up +- FFmpeg error β†’ Job retries once, then fails +- Upload error β†’ Job retries once, then fails +- Partial resolution failure β†’ Job continues, master playlist includes successful resolutions + +### Manual Testing + +**Test Cases**: +1. Upload a small video (10-20MB) and verify all 3 resolutions +2. Upload a large video (100MB+) and verify all 3 resolutions +3. Verify master playlist is accessible via API +4. Test master playlist in video player (HLS.js or similar) +5. Verify adaptive streaming switches between resolutions +6. Test error handling with invalid video +7. Test retry logic with transient error +8. Verify temp files are cleaned up on success +9. Verify temp files are cleaned up on failure + +**Test Script**: +```bash +# 1. Start infrastructure +docker-compose up -d + +# 2. Start API +nx serve api + +# 3. Start Worker +nx serve worker + +# 4. Upload video +curl -X POST http://localhost:3000/test-video \ + -F "file=@test-video.mp4" + +# 5. Get video ID from response +VIDEO_ID="" + +# 6. Check all resolutions in MinIO +# (Use MinIO console or API to verify files) + +# 7. Get master playlist +curl http://localhost:3000/videos/$VIDEO_ID/playlist.m3u8 + +# 8. Test in video player +# (Use HLS.js test player or similar) +``` + +--- + +## Success Criteria + +### Functional Success + +- βœ… Videos are transcoded to all three resolutions (360p, 720p, 1080p) +- βœ… All resolution playlists and segments are uploaded to MinIO +- βœ… Master playlist (`index.m3u8`) is generated correctly +- βœ… Master playlist includes bandwidth tags for adaptive streaming +- βœ… Master playlist is accessible via API endpoint +- βœ… Error handling catches all error types +- βœ… Status is updated to FAILED on errors +- βœ… Temporary files are cleaned up in all scenarios +- βœ… Failed jobs are retried once before being marked as failed +- βœ… Progress reporting accounts for multiple resolutions + +### Technical Success + +- βœ… Multi-resolution transcoding doesn't significantly slow down jobs +- βœ… All temporary files are cleaned up (no disk space leaks) +- βœ… Retry logic prevents infinite retries +- βœ… Error messages are clear and actionable +- βœ… Code follows NestJS patterns and conventions +- βœ… All methods have proper error handling + +### User Experience Success + +- βœ… Video players can access master playlist +- βœ… Adaptive streaming works (switches between resolutions) +- βœ… Users on slow connections get 360p +- βœ… Users on fast connections get 1080p +- βœ… No playback interruptions due to quality switching + +--- + +## Future Enhancements (Out of Scope) + +### Phase 2 Features + +- **Parallel Transcoding**: Transcode multiple resolutions in parallel (faster but more resource-intensive) +- **Dynamic Resolution Selection**: Choose resolutions based on source video resolution +- **Custom Resolution Profiles**: Allow users to specify which resolutions to generate +- **Resolution-Specific Progress**: Track progress for each resolution separately +- **Quality Metrics**: Measure and report quality metrics per resolution +- **Bandwidth Optimization**: Automatically adjust bitrates based on content complexity + +### Optimization + +- **Parallel Transcoding**: Use worker pools to transcode resolutions in parallel +- **Incremental Upload**: Upload segments as they're generated (not after all transcoding) +- **Caching**: Cache transcoded resolutions for frequently accessed videos +- **CDN Integration**: Serve HLS files from CDN for better performance + +--- + +## Dependencies + +### External Dependencies + +- **FFmpeg**: Video transcoding (already installed) +- **fluent-ffmpeg**: Node.js FFmpeg wrapper (already installed) +- **fs-extra**: File system operations (already installed) +- **MinIO**: Object storage (already configured) + +### Internal Dependencies + +- **TranscoderService**: Core transcoding logic +- **StorageService**: MinIO upload operations +- **QueueConsumerService**: Job consumption and error handling +- **QueueService**: Job publishing +- **RedisService**: Progress tracking +- **VideosService**: Status updates + +### Blocking Dependencies + +- None (all dependencies are already available) + +--- + +## Risks & Mitigations + +### Risk 1: Multi-Resolution Transcoding Takes Too Long + +**Risk**: Transcoding three resolutions sequentially may take 3x longer than single resolution. + +**Mitigation**: +- Monitor transcoding times and adjust if needed +- Consider parallel transcoding in future (out of scope) +- Optimize FFmpeg parameters for faster encoding +- Use lower quality presets for 360p and 720p + +### Risk 2: Disk Space Issues + +**Risk**: Multiple resolutions create more temporary files, increasing disk space usage. + +**Mitigation**: +- Ensure cleanup in finally block always runs +- Monitor disk space usage +- Set appropriate temp directory limits +- Clean up temp files immediately after upload + +### Risk 3: Master Playlist Generation Errors + +**Risk**: Master playlist generation fails, leaving videos without adaptive streaming. + +**Mitigation**: +- Generate master playlist after all resolutions complete +- Include only successfully transcoded resolutions +- Log errors but don't fail entire job if master playlist fails +- Fall back to single resolution playlist if needed + +### Risk 4: Retry Logic Causes Infinite Loops + +**Risk**: Retry logic may cause jobs to retry indefinitely. + +**Mitigation**: +- Enforce max retry count (1 retry) +- Track retry count in message headers +- Non-retryable errors should not be retried +- Monitor retry patterns and adjust if needed + +### Risk 5: Progress Reporting Inaccuracy + +**Risk**: Progress reporting may not accurately reflect multi-resolution workflow. + +**Mitigation**: +- Test progress reporting with various video sizes +- Adjust progress milestones based on actual transcoding times +- Monitor and adjust progress calculation formula + +--- + +## Appendix + +### Resolution Configuration Example + +```typescript +const RESOLUTIONS: ResolutionConfig[] = [ + { + height: 360, + width: 640, + bandwidth: 800000, + label: '360p', + }, + { + height: 720, + width: 1280, + bandwidth: 2500000, + label: '720p', + }, + { + height: 1080, + width: 1920, + bandwidth: 5000000, + label: '1080p', + }, +]; +``` + +### Master Playlist Example + +``` +#EXTM3U +#EXT-X-VERSION:3 +#EXT-X-STREAM-INF:BANDWIDTH=800000,RESOLUTION=640x360 +360p/360p.m3u8 +#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=1280x720 +720p/720p.m3u8 +#EXT-X-STREAM-INF:BANDWIDTH=5000000,RESOLUTION=1920x1080 +1080p/1080p.m3u8 +``` + +### Job Data Example + +```json +{ + "videoId": "123e4567-e89b-12d3-a456-426614174000", + "sourceBucket": "raw-videos", + "sourceKey": "123e4567-e89b-12d3-a456-426614174000.mp4", + "targetBucket": "processed-videos", + "resolutions": [ + { + "height": 360, + "width": 640, + "bandwidth": 800000, + "label": "360p" + }, + { + "height": 720, + "width": 1280, + "bandwidth": 2500000, + "label": "720p" + }, + { + "height": 1080, + "width": 1920, + "bandwidth": 5000000, + "label": "1080p" + } + ] +} +``` + +### Error Response Example + +```json +{ + "statusCode": 500, + "message": "Transcoding failed: FFmpeg error", + "error": "Internal Server Error", + "videoId": "123e4567-e89b-12d3-a456-426614174000", + "status": "FAILED", + "retryCount": 1 +} +``` + +--- + +## References + +- [HLS Master Playlist Specification](https://tools.ietf.org/html/rfc8216) +- [FFmpeg Documentation](https://ffmpeg.org/documentation.html) +- [fluent-ffmpeg Documentation](https://github.com/fluent-ffmpeg/node-fluent-ffmpeg) +- Existing PRDs: `docs/done/PRD-REDIS-PROGRESS-TRACKING.md`, `docs/done/PRD-FFMPEG-TRANSCODING.md` +- Transcoder Service: `apps/worker/src/app/transcoder/transcoder.service.ts` +- Queue Consumer Service: `apps/worker/src/app/queue-consumer/queue-consumer.service.ts` +- Queue Service: `apps/api/src/app/queue/queue.service.ts` + diff --git a/docs/active/PRD-REDIS-PROGRESS-TRACKING.md b/docs/done/PRD-REDIS-PROGRESS-TRACKING.md similarity index 100% rename from docs/active/PRD-REDIS-PROGRESS-TRACKING.md rename to docs/done/PRD-REDIS-PROGRESS-TRACKING.md