Feature/rabbitmq integration

.vscode/launch.json

@@ -18,6 +18,24 @@
         "${workspaceFolder}/apps/api/dist/**/*.(m|c|)js",
         "!**/node_modules/**"
       ]
+    },
+    {
+      "type": "node",
+      "request": "launch",
+      "name": "Debug worker with Nx",
+      "runtimeExecutable": "npx",
+      "runtimeArgs": ["nx", "serve", "worker"],
+      "env": {
+        "NODE_OPTIONS": "--inspect=9230"
+      },
+      "console": "integratedTerminal",
+      "internalConsoleOptions": "neverOpen",
+      "skipFiles": ["<node_internals>/**"],
+      "sourceMaps": true,
+      "outFiles": [
+        "${workspaceFolder}/apps/worker/dist/**/*.(m|c|)js",
+        "!**/node_modules/**"
+      ]
     }
   ]
 }

.vscode/tasks.json

@@ -23,6 +23,27 @@
       },
       "problemMatcher": []
     },
+    {
+      "label": "💻 Worker",
+      "type": "shell",
+      "command": "npx",
+      "args": ["nx", "serve", "worker", "--no-tui"],
+      "group": "build",
+      "presentation": {
+        "echo": true,
+        "reveal": "always",
+        "focus": false,
+        "panel": "dedicated",
+        "showReuseMessage": true,
+        "clear": false
+      },
+      "options": {
+        "env": {
+          "NODE_ENV": "development"
+        }
+      },
+      "problemMatcher": []
+    },
     {
       "label": "🛑 Stop Nx Daemon",
       "type": "shell",

apps/api/src/app/app.module.ts

@@ -4,6 +4,7 @@ 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';
 
@@ -31,6 +32,7 @@ import { VideosModule } from './videos/videos.module';
     }),
     VideosModule,
     StorageModule,
+    QueueModule,
   ],
   controllers: [AppController],
   providers: [AppService],

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 {}
+

apps/api/src/app/queue/queue.service.ts

@@ -0,0 +1,192 @@
+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);
+
+  private connection: amqp.ChannelModel | null = null;
+
+  private channel: amqp.Channel | null = null;
+
+  constructor(private readonly configService: ConfigService) {}
+
+  async onModuleInit(): Promise<void> {
+    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<void> {
+    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);
+    }
+  }
+
+  /**
+   * Initialize RabbitMQ connection and channel.
+   * Idempotent: if connection & channel already exist, it returns early.
+   */
+  private async initializeConnection(): Promise<void> {
+    if (this.connection && this.channel) {
+      // Already initialized
+      return;
+    }
+
+    const host = this.configService.get<string>('rabbitmq.host');
+    const port = this.configService.get<number>('rabbitmq.port');
+    const user = this.configService.get<string>('rabbitmq.user');
+    const password = this.configService.get<string>('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;
+    }
+  }
+
+  /**
+   * Declare the durable transcode queue.
+   * Safe to call multiple times (RabbitMQ assertQueue is idempotent).
+   */
+  private async declareQueue(): Promise<void> {
+    if (!this.channel) {
+      // Ensure we have a channel first
+      await this.initializeConnection();
+    }
+
+    const queueName = this.configService.get<string>('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<void> {
+    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;
+    }
+  }
+
+  /**
+   * Publish a transcode job message to the configured RabbitMQ queue.
+   */
+  async publishTranscodeJob(
+    videoId: string,
+    s3Key: string,
+    resolutions: string[],
+  ): Promise<void> {
+    // 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<string>('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;
+    }
+  }
+}
+

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],

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<Video>,
     private storageService: StorageService,
+    private queueService: QueueService,
   ) {}
 
   /**
@@ -55,18 +57,50 @@ export class VideosService {
       await this.storageService.uploadStream(uploadData, s3Key, metadata, fileSize);
       this.logger.log(`File uploaded successfully to MinIO: s3Key=${s3Key}`);
 
-      // Create video record in database
+      // Create video record in database with status UPLOADED
       const video = await this.create(
         originalFilename,
         s3Key,
         mimeType,
         VideoStatus.UPLOADED,
       );
 
-      this.logger.log(
-        `Video upload and record creation completed: id=${video.id}, s3Key=${video.s3Key}`,
-      );
-      return video;
+      // After successful upload and DB record creation, publish transcode job
+      try {
+        const defaultResolutions = ['360p', '720p', '1080p'];
+
+        this.logger.log(
+          `Publishing transcode job for video: id=${video.id}, s3Key=${video.s3Key}, resolutions=${defaultResolutions.join(
+            ',',
+          )}`,
+        );
+
+        await this.queueService.publishTranscodeJob(
+          video.id,
+          s3Key,
+          defaultResolutions,
+        );
+
+        // Update status to QUEUED after successful publish
+        const queuedVideo = await this.updateStatus(
+          video.id,
+          VideoStatus.QUEUED,
+        );
+
+        this.logger.log(
+          `Video upload, record creation, and queue publish completed: id=${queuedVideo.id}, status=${queuedVideo.status}`,
+        );        return queuedVideo;
+      } catch (error) {
+        // If queue publishing fails, log error but do not fail the upload
+        this.logger.error(
+          `Failed to publish transcode job for video id=${video.id}, s3Key=${s3Key}. Keeping status as UPLOADED. Error: ${error.message}`,
+          error.stack,
+        );
+
+        this.logger.log(
+          `Video upload and record creation completed (without queue publish): id=${video.id}, s3Key=${video.s3Key}, status=${video.status}`,
+        );        return video;
+      }
     } catch (error) {
       this.logger.error(
         `Failed to upload video file: ${error.message}`,

apps/worker-e2e/eslint.config.mjs

@@ -0,0 +1,3 @@
+import baseConfig from '../../eslint.config.mjs';
+
+export default [...baseConfig];

apps/worker-e2e/jest.config.cts

@@ -0,0 +1,18 @@
+export default {
+  displayName: 'worker-e2e',
+  preset: '../../jest.preset.js',
+  globalSetup: '<rootDir>/src/support/global-setup.ts',
+  globalTeardown: '<rootDir>/src/support/global-teardown.ts',
+  setupFiles: ['<rootDir>/src/support/test-setup.ts'],
+  testEnvironment: 'node',
+  transform: {
+    '^.+\\.[tj]s$': [
+      'ts-jest',
+      {
+        tsconfig: '<rootDir>/tsconfig.spec.json',
+      },
+    ],
+  },
+  moduleFileExtensions: ['ts', 'js', 'html'],
+  coverageDirectory: '../../coverage/worker-e2e',
+};