From d58072fc11186c06e056462bb64146d764f51d83 Mon Sep 17 00:00:00 2001 From: Giorgi Zankaidze Date: Mon, 15 Dec 2025 01:07:47 +0400 Subject: [PATCH 01/14] add PRD --- docs/PRD/PRD-RABBITMQ-INTEGRATION.md | 1207 ++++++++++++++++++++++++++ 1 file changed, 1207 insertions(+) create mode 100644 docs/PRD/PRD-RABBITMQ-INTEGRATION.md diff --git a/docs/PRD/PRD-RABBITMQ-INTEGRATION.md b/docs/PRD/PRD-RABBITMQ-INTEGRATION.md new file mode 100644 index 0000000..983831d --- /dev/null +++ b/docs/PRD/PRD-RABBITMQ-INTEGRATION.md @@ -0,0 +1,1207 @@ +# Product Requirements Document: RabbitMQ Integration + +## Executive Summary + +This document defines the requirements for implementing RabbitMQ message queue integration for the StreamForge video transcoding platform. The queue service will enable asynchronous job processing by publishing transcoding jobs to a durable message queue, while a separate worker application will consume and process these jobs. This integration is a critical component that decouples the API layer from CPU-intensive transcoding operations, enabling scalable and resilient video processing. + +**Timeline**: Day 8-9 (4-6 hours) +**Effort Estimate**: 4-6 hours +**Priority**: High (Blocking Day 10-11 FFmpeg transcoding implementation) + +--- + +## Background & Context + +### Why RabbitMQ Integration is Needed + +StreamForge requires asynchronous job processing to handle CPU-intensive video transcoding operations. Processing videos synchronously in the API would cause: +- Long HTTP request timeouts (transcoding can take minutes) +- Poor user experience (blocking requests) +- Inability to scale transcoding independently +- Resource contention on API servers + +RabbitMQ provides a message broker that: +- Enables asynchronous job processing +- Supports durable queues (survive restarts) +- Allows horizontal scaling of workers +- Provides message persistence and delivery guarantees +- Offers management UI for monitoring + +### Integration with Existing System + +The queue service integrates with: +- **Video Entity** (`apps/api/src/app/videos/entities/video.entity.ts`): The `status` field tracks job state (`UPLOADED` → `QUEUED` → `PROCESSING`) +- **Videos Service** (`apps/api/src/app/videos/videos.service.ts`): After successful upload, publishes job to queue +- **Configuration** (`apps/api/src/config/configuration.ts`): RabbitMQ connection settings are already defined +- **Docker Infrastructure** (`docker-compose.yml`): RabbitMQ container is already configured and running +- **Future Worker** (Day 10-11): Worker will consume jobs and perform transcoding + +### Architecture Context + +The queue service follows the StreamForge architecture principle: **Workers Do Heavy Lifting**. The NestJS API only receives uploads and queues jobs. It never does heavy processing. This separation enables: +- Fast API response times (upload completes quickly) +- Independent scaling of API and workers +- Fault tolerance (workers can restart without affecting API) +- Support for multiple worker instances processing jobs in parallel + +--- + +## Objectives & Goals + +### Primary Objectives + +1. **Job Publishing**: Enable API to publish transcoding jobs to RabbitMQ queue +2. **Queue Management**: Create and manage durable `transcode_queue` for job persistence +3. **Worker Foundation**: Set up separate worker application to consume jobs +4. **Integration**: Seamlessly integrate queue publishing into video upload workflow +5. **Message Reliability**: Ensure jobs are persisted and not lost on restarts + +### Success Criteria + +- ✅ Queue service successfully connects to RabbitMQ on module initialization +- ✅ `transcode_queue` is declared as durable and persists across restarts +- ✅ Jobs can be published with video metadata (videoId, s3Key, resolutions) +- ✅ Worker application can consume jobs from queue +- ✅ Jobs are logged when received by worker (no processing yet) +- ✅ Video status updates to `QUEUED` after job publishing +- ✅ Service integrates seamlessly with existing NestJS application structure +- ✅ Test: Upload video → Check RabbitMQ UI → See job in queue → Worker logs it + +### Key Deliverables + +- `apps/api/src/app/queue/queue.module.ts` - NestJS module for queue service +- `apps/api/src/app/queue/queue.service.ts` - Core queue service with publishing methods +- `apps/worker/src/main.ts` - Worker application with RabbitMQ consumer +- Updated `apps/api/src/app/videos/videos.service.ts` - Integration with queue service +- Working connection to RabbitMQ with queue declaration +- Test verification: Upload video → Job appears in queue → Worker receives and logs job + +--- + +## Functional Requirements + +### FR1: Queue Module Initialization + +**Requirement**: The queue module must initialize the RabbitMQ connection when the module is loaded. + +**Details**: +- Connection should be established using configuration from `ConfigService` +- Connection parameters: host, port, username, password +- Connection should be validated on initialization +- Connection errors should be logged and handled gracefully +- Connection should be established lazily (on first use) or eagerly (on module init) + +**Implementation Notes**: +- Use `amqplib` package for RabbitMQ connection +- Connection can be established in `onModuleInit()` lifecycle hook +- Store connection and channel as private class properties +- Handle connection failures gracefully (log error, allow retry) + +### FR2: Queue Connection Management + +**Requirement**: The service must manage RabbitMQ connection and channel lifecycle. + +**Details**: +- Establish connection to RabbitMQ broker +- Create and manage channel for publishing messages +- Handle connection errors and reconnection logic (basic implementation) +- Close connection gracefully on module destruction +- Support connection health checks + +**Connection Management**: +- Connection should be established once and reused +- Channel should be created per connection +- Handle connection drops and channel errors +- Log connection state changes + +### FR3: Job Publishing + +**Requirement**: `publishTranscodeJob(videoId, s3Key, resolutions)` method must publish jobs to queue. + +**Method Signature**: +```typescript +publishTranscodeJob( + videoId: string, + s3Key: string, + resolutions: string[] +): Promise +``` + +**Details**: +- Accept video metadata as parameters +- Serialize job data to JSON format +- Publish message to `transcode_queue` +- Ensure message persistence (persistent delivery mode) +- Return void on success +- Handle publishing errors appropriately + +**Parameters**: +- `videoId`: UUID of the video entity +- `s3Key`: MinIO object key where raw video is stored (e.g., `raw-videos/{videoId}.mp4`) +- `resolutions`: Array of resolution strings (e.g., `['360p', '720p', '1080p']`) + +**Message Format**: +```json +{ + "videoId": "123e4567-e89b-12d3-a456-426614174000", + "s3Key": "raw-videos/123e4567-e89b-12d3-a456-426614174000.mp4", + "resolutions": ["360p", "720p", "1080p"] +} +``` + +**Returns**: Promise resolving to void + +**Errors**: Should throw descriptive errors for: +- Connection failures +- Channel errors +- Publishing failures +- Invalid parameters + +### FR4: Queue Declaration + +**Requirement**: The service must declare `transcode_queue` as a durable queue. + +**Details**: +- Declare queue on module initialization or first publish +- Queue name: `transcode_queue` (configurable via environment variable) +- Queue should be durable (survives RabbitMQ restarts) +- Queue should persist messages (survives broker restarts) +- Queue declaration should be idempotent (safe to call multiple times) +- Log queue declaration status + +**Queue Configuration**: +- `durable: true` - Queue survives broker restarts +- `autoDelete: false` - Queue is not deleted when unused +- No message TTL initially (messages persist until consumed) +- No priority queue initially (FIFO processing) + +### FR5: Worker Application Setup + +**Requirement**: A separate NestJS worker application must be created in the monorepo. + +**Details**: +- Generate worker app using Nx: `nx generate @nx/nest:application worker` +- Worker should be a standalone NestJS application +- Worker should have its own `main.ts` entry point +- Worker should connect to RabbitMQ independently +- Worker should be runnable separately from API: `nx serve worker` + +**Application Structure**: +- `apps/worker/src/main.ts` - Entry point +- `apps/worker/project.json` - Nx project configuration +- Worker should bootstrap NestJS application +- Worker should handle graceful shutdown + +### FR6: Worker Consumer Setup + +**Requirement**: Worker must consume jobs from `transcode_queue` and log received jobs. + +**Details**: +- Connect to RabbitMQ using same configuration as API +- Declare and bind to `transcode_queue` +- Consume messages from queue +- Parse JSON message payload +- Log received job details (videoId, s3Key, resolutions) +- Acknowledge messages after successful receipt +- Handle message parsing errors gracefully +- No transcoding processing yet (just logging) + +**Consumer Configuration**: +- `noAck: false` - Manual acknowledgment required +- `prefetch: 1` - Process one job at a time (for now) +- Acknowledge message after logging (not after processing) +- Handle consumer errors and log them + +**Logging Format**: +``` +[Worker] Received transcode job: videoId={videoId}, s3Key={s3Key}, resolutions={resolutions} +``` + +### FR7: Integration with Videos Service + +**Requirement**: Videos service must publish jobs after successful video upload. + +**Details**: +- Inject QueueService into VideosService +- After successful MinIO upload and database record creation +- Call `queueService.publishTranscodeJob(videoId, s3Key, resolutions)` +- Update video status to `QUEUED` after successful job publishing +- Handle queue publishing errors (log error, but don't fail upload) +- Default resolutions: `['360p', '720p', '1080p']` (can be configurable later) + +**Integration Flow**: +1. Video uploaded to MinIO → Status: `UPLOADED` +2. Database record created +3. Job published to queue → Status: `QUEUED` +4. Worker consumes job → Status: `PROCESSING` (future) +5. Transcoding completes → Status: `COMPLETED` (future) + +**Error Handling**: +- If queue publishing fails, log error but don't fail the upload +- Video status remains `UPLOADED` if queue publishing fails +- Consider retry logic for queue publishing (optional for Day 8-9) + +### FR8: Error Handling & Logging + +**Requirement**: All queue operations must handle errors appropriately and log important events. + +**Details**: +- Catch and log RabbitMQ connection errors +- Catch and log channel errors +- Catch and log publishing errors +- Catch and log consumer errors +- Provide descriptive error messages +- Use NestJS Logger for consistent logging +- Log connection establishment +- Log queue declaration +- Log job publishing (with videoId, not full message) +- Log job consumption (with videoId) + +**Error Types**: +- `QueueConnectionError`: RabbitMQ connection failures +- `QueuePublishError`: Job publishing failures +- `QueueConsumeError`: Job consumption failures +- `QueueValidationError`: Invalid parameters + +--- + +## Technical Specifications + +### Technology Stack + +- **RabbitMQ Client Library**: `amqplib` package (v0.10.9, already installed) +- **Node.js**: Native support for async/await +- **NestJS**: Module and service patterns +- **TypeScript**: Full type safety +- **Nx**: Monorepo application generation + +### Configuration Requirements + +The service will use configuration from `apps/api/src/config/configuration.ts`: + +```typescript +rabbitmq: { + host: string, // e.g., 'localhost' + port: number, // e.g., 5672 + user: string, // e.g., 'admin' + password: string, // e.g., 'admin' + queueName: string, // 'transcode_queue' +} +``` + +### Queue Configuration + +**Queue Name**: `transcode_queue` (configurable via `RABBITMQ_QUEUE_NAME`) + +**Queue Properties**: +- `durable: true` - Queue survives broker restarts +- `autoDelete: false` - Queue persists when unused +- No arguments initially (no TTL, priority, etc.) + +**Message Properties**: +- `persistent: true` - Messages survive broker restarts +- `contentType: 'application/json'` - JSON message format +- `deliveryMode: 2` - Persistent delivery mode + +### Message Format + +**Job Message Structure**: +```typescript +interface TranscodeJob { + videoId: string; // UUID of video entity + s3Key: string; // MinIO object key (e.g., 'raw-videos/{videoId}.mp4') + resolutions: string[]; // Array of resolution strings (e.g., ['360p', '720p', '1080p']) +} +``` + +**JSON Serialization**: +- Use `JSON.stringify()` to serialize job object +- Use `JSON.parse()` to deserialize in worker +- Validate message structure in worker before processing + +### RabbitMQ Connection + +**Connection URL Format**: +``` +amqp://{user}:{password}@{host}:{port} +``` + +**Example**: +``` +amqp://admin:admin@localhost:5672 +``` + +**Connection Options**: +- Use `amqplib.connect()` with connection URL +- Handle connection errors +- Support connection retry (basic implementation) + +**Channel Management**: +- Create channel from connection: `connection.createChannel()` +- Handle channel errors +- Reuse channel for multiple publishes +- Close channel on module destruction + +### amqplib Usage Patterns + +**Connection**: +```typescript +import * as amqp from 'amqplib'; + +const connection = await amqp.connect(connectionUrl); +const channel = await connection.createChannel(); +``` + +**Queue Declaration**: +```typescript +await channel.assertQueue(queueName, { durable: true }); +``` + +**Publishing**: +```typescript +const message = JSON.stringify(jobData); +channel.sendToQueue(queueName, Buffer.from(message), { persistent: true }); +``` + +**Consuming**: +```typescript +await channel.consume(queueName, (msg) => { + if (msg) { + const jobData = JSON.parse(msg.content.toString()); + // Process job + channel.ack(msg); + } +}, { noAck: false }); +``` + +--- + +## Architecture & Design + +### Module Structure + +``` +apps/api/src/app/queue/ +├── queue.module.ts # NestJS module definition +└── queue.service.ts # Core queue service implementation +``` + +### Module Definition (`queue.module.ts`) + +```typescript +@Module({ + imports: [ConfigModule], // Already global, but explicit for clarity + providers: [QueueService], + exports: [QueueService], // Export for use in other modules +}) +export class QueueModule {} +``` + +### Service Lifecycle + +1. **Module Initialization**: QueueModule is imported into AppModule +2. **Service Construction**: QueueService is instantiated +3. **OnModuleInit**: Connection to RabbitMQ is established (optional - can be lazy) +4. **Queue Declaration**: `transcode_queue` is declared as durable +5. **Ready State**: Service is ready to publish jobs +6. **OnModuleDestroy**: Connection is closed gracefully + +### Integration Points + +**With AppModule**: +- QueueModule will be imported into `apps/api/src/app/app.module.ts` +- Service will be available for injection in other modules + +**With VideosModule**: +- VideosService will inject QueueService +- After video upload, `publishTranscodeJob()` will be called +- Video status will be updated to `QUEUED` + +**With Configuration**: +- Uses ConfigService to access RabbitMQ configuration +- Configuration is already loaded globally via ConfigModule + +**With Worker Application**: +- Worker connects to same RabbitMQ instance +- Worker consumes from same `transcode_queue` +- Worker and API are decoupled (can run independently) + +### Dependency Injection + +```typescript +@Injectable() +export class QueueService implements OnModuleInit, OnModuleDestroy { + private connection: amqp.Connection; + private channel: amqp.Channel; + + constructor( + private configService: ConfigService, + private logger: Logger, + ) {} + + async onModuleInit() { + await this.initializeConnection(); + await this.declareQueue(); + } + + async onModuleDestroy() { + await this.closeConnection(); + } +} +``` + +### Worker Application Structure + +``` +apps/worker/ +├── src/ +│ └── main.ts # Worker entry point with consumer +├── project.json # Nx project configuration +└── tsconfig.json # TypeScript configuration +``` + +**Worker main.ts Structure**: +```typescript +async function bootstrap() { + // Connect to RabbitMQ + // Declare queue + // Consume messages + // Log received jobs + // Handle graceful shutdown +} + +bootstrap(); +``` + +### Error Handling Strategy + +**Connection Errors**: +- Log error with context +- Throw descriptive exception +- Application startup may fail if RabbitMQ is unavailable (fail-fast approach) +- Or: Lazy connection (connect on first publish) + +**Publishing Errors**: +- Catch channel errors +- Log error with context (videoId, queue name) +- Throw appropriate exception for caller to handle +- Consider retry logic (optional) + +**Consuming Errors**: +- Catch message parsing errors +- Log error and reject message (nack) +- Catch processing errors +- Log error and handle appropriately + +**Error Types**: +- `QueueConnectionError`: RabbitMQ connection failures +- `QueuePublishError`: Job publishing failures +- `QueueConsumeError`: Job consumption failures +- `QueueValidationError`: Invalid parameters + +--- + +## Non-Functional Requirements + +### Performance + +- **Publishing Latency**: Job publishing should complete in < 100ms +- **Connection Reuse**: Connection should be established once and reused +- **Message Size**: Messages are small (JSON, < 1KB), no performance concerns +- **Concurrent Publishing**: Support multiple concurrent job publishes + +### Reliability + +- **Message Persistence**: Jobs must survive RabbitMQ restarts (durable queue, persistent messages) +- **Connection Retry**: Consider retry logic for transient connection errors +- **Error Recovery**: Service should recover from temporary RabbitMQ unavailability +- **Idempotency**: Queue declaration should be idempotent +- **Graceful Shutdown**: Close connections gracefully on application shutdown + +### Security + +- **Credentials Management**: Credentials stored in environment variables, not hardcoded +- **Access Control**: RabbitMQ user should have appropriate permissions +- **Input Validation**: Validate job parameters before publishing +- **Message Validation**: Validate message structure in worker + +### Scalability + +- **Horizontal Scaling**: Multiple API instances can publish to same queue +- **Worker Scaling**: Multiple worker instances can consume from same queue (round-robin) +- **Queue Growth**: Queue can handle thousands of pending jobs +- **Future Load Balancing**: Design supports future worker load balancing + +--- + +## Dependencies + +### External Dependencies + +- **amqplib** (v0.10.9): Already installed in `package.json` + - Provides RabbitMQ client library + - Supports promises and async/await + - Handles connection management + +- **@types/amqplib** (v0.10.8): Already installed in devDependencies + - Provides TypeScript type definitions + +### Internal Dependencies + +- **@nestjs/config**: Already installed and configured + - Provides ConfigService for accessing configuration + - ConfigModule is already global in AppModule + +- **@nestjs/common**: Already installed + - Provides Logger, Injectable, OnModuleInit, OnModuleDestroy decorators + - Provides exception classes + +- **@nx/nest**: Already installed + - Provides Nx generator for creating NestJS applications + - Used to generate worker application + +### Infrastructure Dependencies + +- **RabbitMQ Docker Container**: Already configured in `docker-compose.yml` + - Running on `localhost:5672` (AMQP port) + - Management UI on `localhost:15672` + - Default credentials: `admin`/`admin` + - Must be running before API/Worker starts (or handle connection failures gracefully) + +--- + +## Testing Requirements + +### Unit Testing + +**QueueService Tests**: +- Mock amqplib connection and channel +- Test connection initialization +- Test queue declaration logic +- Test `publishTranscodeJob()` method +- Test error handling scenarios +- Test with various job parameters + +**Test File**: `apps/api/src/app/queue/queue.service.spec.ts` + +### Integration Testing + +**RabbitMQ Integration Tests**: +- Use test RabbitMQ instance or Docker container +- Test actual connection establishment +- Test queue declaration +- Test job publishing +- Test message persistence (restart RabbitMQ, verify queue exists) +- Verify messages appear in RabbitMQ management UI + +**Test Approach**: +- Start RabbitMQ container +- Publish test job +- Verify job appears in queue (via management UI or consumer) +- Consume job and verify content +- Test error scenarios (connection failures, etc.) + +### Manual Verification Steps + +1. **Start Infrastructure**: + ```bash + docker-compose up -d + ``` + +2. **Verify RabbitMQ Running**: + - Check `http://localhost:15672` (RabbitMQ Management UI) + - Login with default credentials (`admin`/`admin`) + - Verify connection is working + +3. **Test Queue Declaration**: + - Start API application + - Check RabbitMQ UI → Queues tab + - Verify `transcode_queue` exists and is durable + +4. **Test Job Publishing**: + - Upload video via API endpoint + - Check RabbitMQ UI → Queues tab → `transcode_queue` + - Verify message count increases + - Click on queue → View messages + - Verify message content matches expected JSON format + +5. **Test Worker Consumption**: + - Start worker application: `nx serve worker` + - Upload video via API endpoint + - Check worker logs for received job message + - Check RabbitMQ UI → Verify message is consumed (removed from queue) + +### Test Data Requirements + +- Test video file for upload (small file, 5-10 MB) +- Various video IDs (UUIDs) +- Different s3Key formats +- Different resolution arrays (empty, single, multiple) + +--- + +## Implementation Details + +### File Structure + +``` +apps/api/src/app/queue/ +├── queue.module.ts # Module definition +├── queue.service.ts # Service implementation +└── queue.service.spec.ts # Unit tests (optional for Day 8-9) + +apps/worker/ +├── src/ +│ └── main.ts # Worker entry point +├── project.json # Nx project configuration +└── tsconfig.json # TypeScript configuration +``` + +### Code Organization + +**queue.module.ts**: +- Module decorator +- Imports ConfigModule +- Provides and exports QueueService + +**queue.service.ts**: +- Injectable service class +- Implements OnModuleInit, OnModuleDestroy +- Private connection and channel instances +- Public method: `publishTranscodeJob()` +- Private helper methods: `initializeConnection()`, `declareQueue()`, `closeConnection()` + +**worker/src/main.ts**: +- Async bootstrap function +- RabbitMQ connection setup +- Queue declaration +- Message consumer setup +- Job logging +- Graceful shutdown handling + +### Key Implementation Patterns + +**Connection Initialization**: +```typescript +async onModuleInit() { + await this.initializeConnection(); + await this.declareQueue(); +} + +private async initializeConnection() { + const host = this.configService.get('rabbitmq.host'); + const port = this.configService.get('rabbitmq.port'); + const user = this.configService.get('rabbitmq.user'); + const password = this.configService.get('rabbitmq.password'); + + const connectionUrl = `amqp://${user}:${password}@${host}:${port}`; + this.connection = await amqp.connect(connectionUrl); + this.channel = await this.connection.createChannel(); +} +``` + +**Queue Declaration**: +```typescript +private async declareQueue() { + const queueName = this.configService.get('rabbitmq.queueName'); + await this.channel.assertQueue(queueName, { durable: true }); + this.logger.log(`Queue declared: ${queueName}`); +} +``` + +**Job Publishing**: +```typescript +async publishTranscodeJob( + videoId: string, + s3Key: string, + resolutions: string[] +): Promise { + const queueName = this.configService.get('rabbitmq.queueName'); + const jobData = { videoId, s3Key, resolutions }; + const message = JSON.stringify(jobData); + + const sent = this.channel.sendToQueue( + queueName, + Buffer.from(message), + { persistent: true } + ); + + if (!sent) { + throw new QueuePublishError(`Failed to publish job for video ${videoId}`); + } + + this.logger.log(`Published transcode job: videoId=${videoId}`); +} +``` + +**Worker Consumer**: +```typescript +async function consumeJobs() { + const channel = await connection.createChannel(); + const queueName = 'transcode_queue'; + + await channel.assertQueue(queueName, { durable: true }); + + await channel.consume(queueName, (msg) => { + if (msg) { + try { + const jobData = JSON.parse(msg.content.toString()); + console.log(`[Worker] Received transcode job:`, jobData); + channel.ack(msg); + } catch (error) { + console.error(`[Worker] Failed to parse job:`, error); + channel.nack(msg, false, false); // Reject and don't requeue + } + } + }, { noAck: false }); +} +``` + +### Configuration Usage + +Access configuration via ConfigService: +```typescript +const host = this.configService.get('rabbitmq.host'); +const port = this.configService.get('rabbitmq.port'); +const user = this.configService.get('rabbitmq.user'); +const password = this.configService.get('rabbitmq.password'); +const queueName = this.configService.get('rabbitmq.queueName'); +``` + +--- + +## Risks & Considerations + +### Connection Failures + +**Risk**: RabbitMQ container may not be running when API/Worker starts. + +**Mitigation**: +- Handle connection errors gracefully +- Log clear error messages +- Consider lazy connection (connect on first use) +- Document requirement for RabbitMQ to be running +- Consider health check endpoint + +### Message Loss + +**Risk**: Messages may be lost if queue is not durable or messages are not persistent. + +**Mitigation**: +- Ensure queue is declared with `durable: true` +- Ensure messages are published with `persistent: true` +- Test message persistence (restart RabbitMQ, verify messages survive) + +### Message Parsing Errors + +**Risk**: Worker may receive malformed messages or fail to parse JSON. + +**Mitigation**: +- Validate message structure in worker +- Handle JSON parsing errors gracefully +- Reject malformed messages (nack, don't requeue) +- Log parsing errors for debugging + +### Queue Declaration Failures + +**Risk**: Queue declaration may fail due to permissions or RabbitMQ issues. + +**Mitigation**: +- Handle "queue already exists" errors gracefully (idempotent) +- Log warnings for expected errors +- Throw exceptions for unexpected errors +- Verify RabbitMQ user has appropriate permissions + +### Worker Consumption Errors + +**Risk**: Worker may crash while processing jobs, losing messages. + +**Mitigation**: +- Acknowledge messages only after successful processing (for future transcoding) +- For Day 8-9: Acknowledge after logging (safe) +- Handle consumer errors and log them +- Consider message TTL for failed jobs (future) + +### Concurrent Publishing + +**Risk**: Multiple API instances publishing concurrently may cause issues. + +**Mitigation**: +- RabbitMQ handles concurrent publishes natively +- Channel is thread-safe for publishing +- No additional locking required + +### Worker Scaling + +**Risk**: Multiple workers may process same job or cause load balancing issues. + +**Mitigation**: +- RabbitMQ handles round-robin distribution automatically +- Each message is delivered to one consumer only +- No additional coordination required + +--- + +## Acceptance Criteria + +### Definition of Done + +The RabbitMQ Integration implementation is complete when: + +1. ✅ **Module Created**: `queue.module.ts` exists and is properly structured +2. ✅ **Service Created**: `queue.service.ts` implements `publishTranscodeJob()` method +3. ✅ **Connection Established**: Service connects to RabbitMQ on module initialization +4. ✅ **Queue Declared**: `transcode_queue` is declared as durable +5. ✅ **Publishing Works**: `publishTranscodeJob()` successfully publishes jobs +6. ✅ **Worker Created**: Worker application exists and can be run independently +7. ✅ **Worker Connects**: Worker connects to RabbitMQ successfully +8. ✅ **Worker Consumes**: Worker consumes jobs from `transcode_queue` +9. ✅ **Worker Logs**: Worker logs received job details +10. ✅ **Integration Works**: Videos service publishes jobs after upload +11. ✅ **Status Updates**: Video status updates to `QUEUED` after publishing +12. ✅ **Error Handling**: All methods handle errors appropriately +13. ✅ **Logging**: Important operations are logged +14. ✅ **Integration**: Module is imported into AppModule +15. ✅ **Testing**: Upload video → Job appears in queue → Worker logs it + +### Verification Checklist + +**Setup Verification**: +- [ ] RabbitMQ Docker container is running (`docker-compose up -d`) +- [ ] RabbitMQ management UI accessible at `http://localhost:15672` +- [ ] API application starts without errors +- [ ] Queue module loads successfully +- [ ] Worker application can be started: `nx serve worker` + +**Queue Verification**: +- [ ] `transcode_queue` exists in RabbitMQ management UI +- [ ] Queue is marked as durable in management UI +- [ ] Queue persists after RabbitMQ restart +- [ ] Queue declaration is idempotent (safe to call multiple times) + +**Publishing Verification**: +- [ ] Job can be published via `publishTranscodeJob()` +- [ ] Published job appears in RabbitMQ management UI +- [ ] Message content matches expected JSON format +- [ ] Message is marked as persistent +- [ ] Multiple jobs can be published concurrently + +**Worker Verification**: +- [ ] Worker connects to RabbitMQ successfully +- [ ] Worker declares and binds to `transcode_queue` +- [ ] Worker consumes jobs from queue +- [ ] Worker logs received job details correctly +- [ ] Jobs are removed from queue after consumption +- [ ] Worker handles message parsing errors gracefully + +**Integration Verification**: +- [ ] QueueModule is imported in AppModule +- [ ] QueueService can be injected in VideosService +- [ ] Videos service publishes job after upload +- [ ] Video status updates to `QUEUED` after publishing +- [ ] Upload → Queue → Worker flow works end-to-end + +**Error Handling Verification**: +- [ ] Connection errors are logged appropriately +- [ ] Publishing errors are handled gracefully +- [ ] Worker handles malformed messages correctly +- [ ] Service handles RabbitMQ unavailability gracefully + +**Performance Verification**: +- [ ] Job publishing completes in < 100ms +- [ ] Multiple concurrent publishes work correctly +- [ ] Worker can consume jobs efficiently + +--- + +## Architecture Diagram + +```mermaid +graph TB + AppModule[AppModule] --> QueueModule[QueueModule] + QueueModule --> QueueService[QueueService] + QueueService --> ConfigService[ConfigService] + QueueService --> RabbitMQConnection[RabbitMQ Connection] + RabbitMQConnection --> RabbitMQContainer[RabbitMQ Container
localhost:5672] + + QueueService --> TranscodeQueue[transcode_queue
Durable Queue] + + VideosModule[VideosModule] -.->|Uses| QueueService + VideosService[VideosService] -->|Publishes Jobs| QueueService + + WorkerApp[Worker Application
apps/worker] --> WorkerConnection[RabbitMQ Connection] + WorkerConnection --> RabbitMQContainer + WorkerConnection -->|Consumes| TranscodeQueue + + UploadFlow[Video Upload] --> VideosService + VideosService -->|1. Upload to MinIO| MinIO[MinIO Storage] + VideosService -->|2. Create DB Record| PostgreSQL[PostgreSQL] + VideosService -->|3. Publish Job| QueueService + QueueService -->|4. Job in Queue| TranscodeQueue + TranscodeQueue -->|5. Worker Consumes| WorkerApp + WorkerApp -->|6. Log Job| Logs[Worker Logs] + + style QueueService fill:#e1f5ff + style RabbitMQContainer fill:#fff4e1 + style TranscodeQueue fill:#e8f5e9 + style WorkerApp fill:#f3e5f5 +``` + +--- + +## Message Flow Diagram + +```mermaid +sequenceDiagram + participant Client + participant API as API/VideosService + participant Queue as QueueService + participant RMQ as RabbitMQ + participant Worker as Worker App + + Client->>API: POST /test-video (upload) + API->>API: Upload to MinIO + API->>API: Create DB record (status: UPLOADED) + API->>Queue: publishTranscodeJob(videoId, s3Key, resolutions) + Queue->>RMQ: Declare transcode_queue (durable) + Queue->>RMQ: Publish JSON message (persistent) + RMQ-->>Queue: Message published + Queue-->>API: Job published + API->>API: Update status to QUEUED + API-->>Client: 201 Created (video record) + + Note over RMQ: Message in queue + + Worker->>RMQ: Connect and consume from transcode_queue + RMQ->>Worker: Deliver message + Worker->>Worker: Parse JSON message + Worker->>Worker: Log job details + Worker->>RMQ: Acknowledge message + RMQ->>RMQ: Remove message from queue + + Note over Worker: Job logged (no processing yet) +``` + +--- + +## Next Steps (Day 10-11 Integration) + +After completing this PRD implementation, the queue integration will be used in Day 10-11 for: + +1. **FFmpeg Transcoding**: Worker will process jobs and transcode videos +2. **Status Updates**: Worker will update video status to `PROCESSING` → `COMPLETED`/`FAILED` +3. **Progress Tracking**: Worker will report progress to Redis (Day 12-13) +4. **Error Handling**: Worker will handle transcoding failures and update status + +The queue service is designed to be a foundational component that enables asynchronous video processing, supporting the entire transcoding pipeline. + +--- + +## Document Version + +**Version**: 1.0 +**Date**: 2024 +**Author**: StreamForge Development Team +**Status**: Draft for Review + +--- + +## Manual Testing Points + +### Test Point 1: Queue Declaration and Module Initialization + +**Objective**: Verify that the queue is automatically declared when the API starts and the queue module initializes correctly. + +**Steps**: +1. Ensure RabbitMQ container is running: `docker-compose up -d` +2. Delete `transcode_queue` from RabbitMQ Management UI (if it exists) +3. Start the NestJS API application: `nx serve api` (or equivalent) +4. Check the application logs for queue declaration messages +5. Open RabbitMQ Management UI at `http://localhost:15672` and verify: + - Login with default credentials (`admin`/`admin`) + - Navigate to Queues tab + - Verify `transcode_queue` exists + - Verify queue is marked as "Durable" + - Queue should be empty (0 messages ready) +6. Restart the API application and verify no errors occur (idempotent queue declaration) + +**Expected Result**: +- API starts successfully without errors +- `transcode_queue` is created automatically +- Logs show queue declaration messages +- Queue is marked as durable in management UI +- Restarting the API does not cause errors (queue already exists) + +--- + +### Test Point 2: Job Publishing Workflow + +**Objective**: Verify that jobs can be published to the queue after video upload and messages appear correctly in RabbitMQ. + +**Steps**: +1. Ensure RabbitMQ container and API are running +2. Upload a test video file via API endpoint: `POST /test-video` + - Use Postman, curl, or similar tool + - Upload a small test video file (5-10 MB) +3. Check API logs for job publishing messages +4. Open RabbitMQ Management UI: + - Navigate to Queues tab + - Click on `transcode_queue` + - Verify message count increased (should show 1 message ready) +5. Click "Get messages" to view the message: + - Verify message payload is valid JSON + - Verify message contains: `videoId`, `s3Key`, `resolutions` + - Verify `videoId` matches the uploaded video ID + - Verify `s3Key` matches the MinIO object key + - Verify `resolutions` is an array (e.g., `["360p", "720p", "1080p"]`) +6. Check database to verify video status is `QUEUED` +7. Upload multiple videos and verify multiple messages in queue + +**Expected Result**: +- Video upload completes successfully +- Job is published to queue +- Message appears in RabbitMQ management UI +- Message content matches expected JSON format +- Video status in database is `QUEUED` +- Multiple jobs can be published successfully + +--- + +### Test Point 3: Worker Consumption and Logging + +**Objective**: Verify that the worker application can consume jobs from the queue and log received job details. + +**Steps**: +1. Ensure RabbitMQ container is running +2. Ensure API is running and has published at least one job to queue +3. Start the worker application in a separate terminal: `nx serve worker` +4. Check worker logs for connection messages +5. Verify worker connects to RabbitMQ successfully +6. Check worker logs for job consumption messages: + - Should see log: `[Worker] Received transcode job: videoId={...}, s3Key={...}, resolutions={...}` + - Verify logged videoId matches published job + - Verify logged s3Key matches published job + - Verify logged resolutions match published job +7. Check RabbitMQ Management UI: + - Navigate to Queues tab → `transcode_queue` + - Verify message count decreased (message was consumed) + - Verify message was removed from queue +8. Publish another job from API and verify worker consumes it immediately +9. Stop worker and publish a job, then restart worker: + - Verify job is still in queue (persistent) + - Verify worker consumes the job after restart + +**Expected Result**: +- Worker connects to RabbitMQ successfully +- Worker consumes jobs from queue +- Worker logs job details correctly +- Jobs are removed from queue after consumption +- Jobs persist in queue if worker is stopped +- Worker consumes persisted jobs after restart + +--- + +### Test Point 4: Error Handling and Edge Cases + +**Objective**: Verify that the queue service and worker handle errors gracefully and validate inputs correctly. + +**Steps**: +1. **Test Connection Failure**: + - Stop RabbitMQ container: `docker-compose stop rabbitmq` + - Try to start API application + - Verify appropriate error is logged + - Verify API handles connection failure gracefully (or fails fast with clear error) + +2. **Test Queue Publishing with Invalid Parameters**: + - Try to publish job with empty videoId + - Try to publish job with invalid s3Key + - Verify appropriate validation errors are thrown + +3. **Test Worker with Malformed Message**: + - Manually publish a malformed message to queue via RabbitMQ Management UI + - Start worker and verify it handles parsing error gracefully + - Verify malformed message is rejected (nack, not requeued) + - Verify error is logged + +4. **Test Multiple Workers**: + - Start two worker instances simultaneously + - Publish multiple jobs (5-10 jobs) + - Verify jobs are distributed between workers (round-robin) + - Verify each job is consumed by only one worker + - Verify all jobs are eventually consumed + +5. **Test Queue Persistence**: + - Publish a job to queue + - Restart RabbitMQ container: `docker-compose restart rabbitmq` + - Verify queue still exists after restart + - Verify message is still in queue (persistent) + - Start worker and verify it consumes the persisted message + +**Expected Result**: +- Connection failures are handled gracefully with clear error messages +- Invalid parameters are validated and appropriate errors are thrown +- Malformed messages are rejected and logged +- Multiple workers can consume jobs in parallel +- Queue and messages persist across RabbitMQ restarts +- All errors are logged with context + +--- + +### Test Point 5: End-to-End Integration Flow + +**Objective**: Verify the complete flow from video upload to worker consumption works correctly. + +**Steps**: +1. Ensure all infrastructure is running: `docker-compose up -d` +2. Start API application: `nx serve api` +3. Start worker application: `nx serve worker` (separate terminal) +4. Upload a video file via API: `POST /test-video` +5. Monitor the complete flow: + - **Step 1**: Check API logs - video uploaded to MinIO + - **Step 2**: Check database - video record created with status `UPLOADED` + - **Step 3**: Check API logs - job published to queue + - **Step 4**: Check database - video status updated to `QUEUED` + - **Step 5**: Check RabbitMQ UI - message in queue + - **Step 6**: Check worker logs - job received and logged + - **Step 7**: Check RabbitMQ UI - message consumed (removed from queue) +6. Verify no errors occurred in any step +7. Upload multiple videos and verify all are processed correctly + +**Expected Result**: +- Complete flow works end-to-end without errors +- All steps complete successfully +- Video status transitions correctly: `UPLOADED` → `QUEUED` +- Job appears in queue and is consumed by worker +- Worker logs job details correctly +- Multiple videos can be processed in sequence +- System handles concurrent uploads correctly + +--- + +## Additional Notes + +### RabbitMQ Management UI Access + +- **URL**: `http://localhost:15672` +- **Default Username**: `admin` +- **Default Password**: `admin` +- **Useful Features**: + - Queues tab: View all queues and message counts + - Queue details: View messages, inspect content + - Connections tab: View active connections + - Channels tab: View active channels + +### Development Tips + +- **Queue Inspection**: Use RabbitMQ Management UI to inspect queue contents during development +- **Message Testing**: Manually publish test messages via Management UI for testing +- **Connection Debugging**: Check Connections tab in Management UI to verify API/Worker connections +- **Log Monitoring**: Monitor both API and Worker logs simultaneously during testing + +### Common Issues + +- **Connection Refused**: Ensure RabbitMQ container is running (`docker-compose ps`) +- **Queue Not Found**: Ensure queue is declared before consuming (worker should declare queue) +- **Messages Not Persisting**: Verify queue is durable and messages are published with `persistent: true` +- **Worker Not Consuming**: Verify worker is connected and bound to correct queue name + From d7a859b5beceec6032be8b770d2dc19ea54ed46e Mon Sep 17 00:00:00 2001 From: Giorgi Zankaidze Date: Mon, 15 Dec 2025 22:10:21 +0400 Subject: [PATCH 02/14] add steps doc --- docs/PRD/PRD-RABBITMQ-INTEGRATION.md | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/docs/PRD/PRD-RABBITMQ-INTEGRATION.md b/docs/PRD/PRD-RABBITMQ-INTEGRATION.md index 983831d..12c6c74 100644 --- a/docs/PRD/PRD-RABBITMQ-INTEGRATION.md +++ b/docs/PRD/PRD-RABBITMQ-INTEGRATION.md @@ -4,6 +4,23 @@ This document defines the requirements for implementing RabbitMQ message queue integration for the StreamForge video transcoding platform. The queue service will enable asynchronous job processing by publishing transcoding jobs to a durable message queue, while a separate worker application will consume and process these jobs. This integration is a critical component that decouples the API layer from CPU-intensive transcoding operations, enabling scalable and resilient video processing. +At a high level, the implementation will be completed in the following steps: + +1. **Create Queue module & service in the API** + Define `queue.module.ts` and `queue.service.ts`, wire `ConfigService`, and set up basic logging. +2. **Implement RabbitMQ connection & queue declaration** + Manage the RabbitMQ connection/channel lifecycle and declare a durable `transcode_queue` on init. +3. **Implement `publishTranscodeJob()` in `QueueService`** + Validate inputs, serialize job data to JSON, and publish persistent messages to the queue with proper error handling. +4. **Integrate `QueueService` into `VideosService`** + After successful upload and DB save, publish a transcode job and update video status to `QUEUED`. +5. **Generate and scaffold the worker application** + Create an Nx NestJS worker app (`apps/worker`) with its own `main.ts` and configuration for RabbitMQ. +6. **Implement worker consumer logic** + Connect to RabbitMQ, consume jobs from `transcode_queue`, parse JSON, log job details, and acknowledge messages. +7. **Test and verify end-to-end** + Use unit tests and manual flows (upload → queue → worker logs) to ensure reliability, error handling, and correct status transitions. + **Timeline**: Day 8-9 (4-6 hours) **Effort Estimate**: 4-6 hours **Priority**: High (Blocking Day 10-11 FFmpeg transcoding implementation) From 9c9d0f89d84fc70c4bfaa123b0a6f34c0ad8cbf8 Mon Sep 17 00:00:00 2001 From: Giorgi Zankaidze Date: Mon, 15 Dec 2025 22:15:45 +0400 Subject: [PATCH 03/14] feat: add QueueModule and QueueService for RabbitMQ integration --- apps/api/src/app/app.module.ts | 2 ++ apps/api/src/app/queue/queue.module.ts | 11 +++++++ apps/api/src/app/queue/queue.service.ts | 39 +++++++++++++++++++++++++ 3 files changed, 52 insertions(+) create mode 100644 apps/api/src/app/queue/queue.module.ts create mode 100644 apps/api/src/app/queue/queue.service.ts diff --git a/apps/api/src/app/app.module.ts b/apps/api/src/app/app.module.ts index d42bc2c..6f1a230 100644 --- a/apps/api/src/app/app.module.ts +++ b/apps/api/src/app/app.module.ts @@ -6,6 +6,7 @@ import { AppController } from './app.controller'; import { AppService } from './app.service'; import { StorageModule } from './storage/storage.module'; import { VideosModule } from './videos/videos.module'; +import { QueueModule } from './queue/queue.module'; @Module({ imports: [ @@ -31,6 +32,7 @@ import { VideosModule } from './videos/videos.module'; }), VideosModule, StorageModule, + QueueModule, ], controllers: [AppController], providers: [AppService], diff --git a/apps/api/src/app/queue/queue.module.ts b/apps/api/src/app/queue/queue.module.ts new file mode 100644 index 0000000..bff277e --- /dev/null +++ b/apps/api/src/app/queue/queue.module.ts @@ -0,0 +1,11 @@ +import { Module } from '@nestjs/common'; +import { ConfigModule } from '@nestjs/config'; +import { QueueService } from './queue.service'; + +@Module({ + imports: [ConfigModule], + providers: [QueueService], + exports: [QueueService], +}) +export class QueueModule {} + diff --git a/apps/api/src/app/queue/queue.service.ts b/apps/api/src/app/queue/queue.service.ts new file mode 100644 index 0000000..8d71589 --- /dev/null +++ b/apps/api/src/app/queue/queue.service.ts @@ -0,0 +1,39 @@ +import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import * as amqp from 'amqplib'; + +@Injectable() +export class QueueService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(QueueService.name); + + // RabbitMQ connection and channel placeholders – will be used in later steps + private connection: amqp.Connection | null = null; + + private channel: amqp.Channel | null = null; + + constructor(private readonly configService: ConfigService) {} + + async onModuleInit(): Promise { + this.logger.log('QueueService initialized. RabbitMQ connection setup will be implemented in later steps.'); + // TODO: initializeConnection(); + // TODO: declareQueue(); + } + + async onModuleDestroy(): Promise { + this.logger.log('QueueService shutting down. RabbitMQ connection cleanup will be implemented in later steps.'); + // TODO: closeConnection(); + } + + // TODO: private async initializeConnection(): Promise {} + + // TODO: private async declareQueue(): Promise {} + + // TODO: private async closeConnection(): Promise {} + + // TODO: async publishTranscodeJob( + // videoId: string, + // s3Key: string, + // resolutions: string[], + // ): Promise {} +} + From 70b113ff807ceee5f19d7c639ba566d05dd714b0 Mon Sep 17 00:00:00 2001 From: Giorgi Zankaidze Date: Mon, 15 Dec 2025 22:16:06 +0400 Subject: [PATCH 04/14] refactor: remove duplicate import of QueueModule in app.module.ts --- apps/api/src/app/app.module.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/api/src/app/app.module.ts b/apps/api/src/app/app.module.ts index 6f1a230..f528583 100644 --- a/apps/api/src/app/app.module.ts +++ b/apps/api/src/app/app.module.ts @@ -4,9 +4,9 @@ import { TypeOrmModule } from '@nestjs/typeorm'; import configuration from '../config/configuration'; import { AppController } from './app.controller'; import { AppService } from './app.service'; +import { QueueModule } from './queue/queue.module'; import { StorageModule } from './storage/storage.module'; import { VideosModule } from './videos/videos.module'; -import { QueueModule } from './queue/queue.module'; @Module({ imports: [ From 724f629d85a9967c07fed68994cce69bc6714c6a Mon Sep 17 00:00:00 2001 From: Giorgi Zankaidze Date: Mon, 15 Dec 2025 22:34:49 +0400 Subject: [PATCH 05/14] feat: implement RabbitMQ connection management in QueueService - Added methods to initialize and close RabbitMQ connections and channels. - Implemented queue declaration with error handling. - Enhanced logging for connection lifecycle events. --- apps/api/src/app/queue/queue.service.ts | 108 +++++++++++++++++++++--- 1 file changed, 98 insertions(+), 10 deletions(-) diff --git a/apps/api/src/app/queue/queue.service.ts b/apps/api/src/app/queue/queue.service.ts index 8d71589..2d04f36 100644 --- a/apps/api/src/app/queue/queue.service.ts +++ b/apps/api/src/app/queue/queue.service.ts @@ -6,29 +6,117 @@ import * as amqp from 'amqplib'; export class QueueService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(QueueService.name); - // RabbitMQ connection and channel placeholders – will be used in later steps - private connection: amqp.Connection | null = null; + private connection: amqp.ChannelModel | null = null; private channel: amqp.Channel | null = null; constructor(private readonly configService: ConfigService) {} async onModuleInit(): Promise { - this.logger.log('QueueService initialized. RabbitMQ connection setup will be implemented in later steps.'); - // TODO: initializeConnection(); - // TODO: declareQueue(); + this.logger.log('Initializing QueueService and establishing RabbitMQ connection...'); + + try { + await this.initializeConnection(); + await this.declareQueue(); + this.logger.log('QueueService initialized successfully. RabbitMQ connection and queue are ready.'); + } catch (error) { + this.logger.error('Failed to initialize QueueService (RabbitMQ connection/queue).', error as Error); + throw error; + } } async onModuleDestroy(): Promise { - this.logger.log('QueueService shutting down. RabbitMQ connection cleanup will be implemented in later steps.'); - // TODO: closeConnection(); + this.logger.log('Shutting down QueueService and closing RabbitMQ connection...'); + + try { + await this.closeConnection(); + this.logger.log('QueueService shut down cleanly. RabbitMQ connection closed.'); + } catch (error) { + this.logger.error('Error while closing RabbitMQ connection in QueueService.', error as Error); + } } - // TODO: private async initializeConnection(): Promise {} + /** + * Initialize RabbitMQ connection and channel. + * Idempotent: if connection & channel already exist, it returns early. + */ + private async initializeConnection(): Promise { + if (this.connection && this.channel) { + // Already initialized + return; + } + + const host = this.configService.get('rabbitmq.host'); + const port = this.configService.get('rabbitmq.port'); + const user = this.configService.get('rabbitmq.user'); + const password = this.configService.get('rabbitmq.password'); + + const connectionUrl = `amqp://${user}:${password}@${host}:${port}`; + + try { + this.logger.log(`Connecting to RabbitMQ at ${connectionUrl}...`); + this.connection = await amqp.connect(connectionUrl); + this.channel = await this.connection.createChannel(); + this.logger.log('RabbitMQ connection and channel established.'); + } catch (error) { + this.logger.error('Failed to establish RabbitMQ connection/channel.', error as Error); + // Clean up partial state + this.connection = null; + this.channel = null; + throw error; + } + } - // TODO: private async declareQueue(): Promise {} + /** + * Declare the durable transcode queue. + * Safe to call multiple times (RabbitMQ assertQueue is idempotent). + */ + private async declareQueue(): Promise { + if (!this.channel) { + // Ensure we have a channel first + await this.initializeConnection(); + } - // TODO: private async closeConnection(): Promise {} + const queueName = this.configService.get('rabbitmq.queueName'); + + try { + this.logger.log(`Declaring RabbitMQ queue '${queueName}' (durable)...`); + await this.channel.assertQueue(queueName, { + durable: true, + }); + this.logger.log(`RabbitMQ queue '${queueName}' declared (durable).`); + } catch (error) { + this.logger.error(`Failed to declare RabbitMQ queue '${queueName}'.`, error as Error); + throw error; + } + } + + /** + * Close RabbitMQ channel and connection gracefully. + */ + private async closeConnection(): Promise { + try { + if (this.channel) { + await this.channel.close(); + this.logger.log('RabbitMQ channel closed.'); + } + } catch (error) { + this.logger.error('Error while closing RabbitMQ channel.', error as Error); + } finally { + this.channel = null; + } + + try { + if (this.connection) { + await this.connection.close(); + this.logger.log('RabbitMQ connection closed.'); + } + } catch (error) { + this.logger.error('Error while closing RabbitMQ connection.', error as Error); + } finally { + this.connection = null; + } + } // TODO: async publishTranscodeJob( // videoId: string, From 8ab730469fcc328d233ecbe2aef22095457ba407 Mon Sep 17 00:00:00 2001 From: Giorgi Zankaidze Date: Mon, 15 Dec 2025 22:47:54 +0400 Subject: [PATCH 06/14] feat: implement publishTranscodeJob method in QueueService - Added the publishTranscodeJob method to publish transcode job messages to RabbitMQ. - Included input validation for videoId, s3Key, and resolutions. - Enhanced error handling and logging for message publishing process. --- apps/api/src/app/queue/queue.service.ts | 75 +++++++++++++++++++++++-- 1 file changed, 70 insertions(+), 5 deletions(-) diff --git a/apps/api/src/app/queue/queue.service.ts b/apps/api/src/app/queue/queue.service.ts index 2d04f36..5ce8574 100644 --- a/apps/api/src/app/queue/queue.service.ts +++ b/apps/api/src/app/queue/queue.service.ts @@ -118,10 +118,75 @@ export class QueueService implements OnModuleInit, OnModuleDestroy { } } - // TODO: async publishTranscodeJob( - // videoId: string, - // s3Key: string, - // resolutions: string[], - // ): Promise {} + /** + * Publish a transcode job message to the configured RabbitMQ queue. + */ + async publishTranscodeJob( + videoId: string, + s3Key: string, + resolutions: string[], + ): Promise { + // Basic input validation + if (!videoId || typeof videoId !== 'string') { + this.logger.warn('publishTranscodeJob called with invalid videoId.'); + throw new Error('Invalid videoId for publishTranscodeJob.'); + } + + if (!s3Key || typeof s3Key !== 'string') { + this.logger.warn('publishTranscodeJob called with invalid s3Key.'); + throw new Error('Invalid s3Key for publishTranscodeJob.'); + } + + if (!Array.isArray(resolutions)) { + this.logger.warn('publishTranscodeJob called with invalid resolutions (must be an array).'); + throw new Error('Invalid resolutions for publishTranscodeJob. Expected an array of strings.'); + } + + const queueName = this.configService.get('rabbitmq.queueName'); + + const jobPayload = { + videoId, + s3Key, + resolutions, + }; + + const message = JSON.stringify(jobPayload); + + try { + // Ensure connection/channel are ready + await this.initializeConnection(); + + if (!this.channel) { + this.logger.error('RabbitMQ channel is not available in publishTranscodeJob.'); + throw new Error('RabbitMQ channel is not available.'); + } + + const sent = this.channel.sendToQueue( + queueName, + Buffer.from(message), + { + persistent: true, + contentType: 'application/json', + }, + ); + + if (!sent) { + this.logger.error( + `Failed to publish transcode job to queue '${queueName}' (sendToQueue returned false). videoId=${videoId}`, + ); + throw new Error(`Failed to publish transcode job for videoId=${videoId}.`); + } + + this.logger.log( + `Published transcode job to queue '${queueName}': videoId=${videoId}`, + ); + } catch (error) { + this.logger.error( + `Error while publishing transcode job to queue '${queueName}' for videoId=${videoId}.`, + error as Error, + ); + throw error; + } + } } From 1f8b920a4a176c556afe73777e893ff52b46cf7a Mon Sep 17 00:00:00 2001 From: Giorgi Zankaidze Date: Mon, 15 Dec 2025 22:59:10 +0400 Subject: [PATCH 07/14] feat: integrate QueueService into VideosService for transcode job publishing - Imported QueueModule into VideosModule to enable QueueService usage. - Updated VideosService to inject QueueService and publish transcode jobs after video upload. - Enhanced logging for video upload, record creation, and queue publishing processes, including error handling for queue publishing failures. --- apps/api/src/app/videos/videos.module.ts | 2 + apps/api/src/app/videos/videos.service.ts | 48 ++++++++++++++++++++--- 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/apps/api/src/app/videos/videos.module.ts b/apps/api/src/app/videos/videos.module.ts index dedcef5..f87e9c4 100644 --- a/apps/api/src/app/videos/videos.module.ts +++ b/apps/api/src/app/videos/videos.module.ts @@ -1,5 +1,6 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; +import { QueueModule } from '../queue/queue.module'; import { StorageModule } from '../storage/storage.module'; import { Video } from './entities/video.entity'; import { VideosService } from './videos.service'; @@ -8,6 +9,7 @@ import { VideosService } from './videos.service'; imports: [ TypeOrmModule.forFeature([Video]), StorageModule, // Import StorageModule to use StorageService + QueueModule, // Import QueueModule to use QueueService ], providers: [VideosService], exports: [VideosService], diff --git a/apps/api/src/app/videos/videos.service.ts b/apps/api/src/app/videos/videos.service.ts index ac1306c..557a871 100644 --- a/apps/api/src/app/videos/videos.service.ts +++ b/apps/api/src/app/videos/videos.service.ts @@ -3,6 +3,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 { StorageService } from '../storage/storage.service'; import { Video, VideoStatus } from './entities/video.entity'; @@ -14,6 +15,7 @@ export class VideosService { @InjectRepository(Video) private videoRepository: Repository