REFACTOR transcription microservice into the monollith

.env.example

@@ -23,5 +23,3 @@ AWS_SECRET_ACCESS_KEY=
 AWS_ACCESS_KEY_ID=
 AWS_REGION=
 AWS_BUCKET=
-
-LAVINMQ_HOST=

.eslintrc.cjs

@@ -44,5 +44,8 @@ module.exports = {
     'import/no-relative-packages': 'off',
     'no-case-declarations': 'off',
     'max-len': 'off',
+    'no-plusplus': 'off',
+    'no-await-in-loop': 'off',
+    'no-restricted-syntax': 'off',
   },
 };

.kilocode/rules/memory-bank/tech.md

@@ -107,9 +107,6 @@ TELEGRAM_BOT_API_KEY=your-bot-token
 
 # Frontend
 FRONT_WEB_APP_ORIGIN_URL=http://localhost:5173
-
-# Message Queue
-LAVINMQ_HOST=localhost
 ```
 
 ### Local Development Commands

compose.yaml

@@ -22,7 +22,6 @@ services:
       AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
       AWS_BUCKET: dev-analysis-entry-storage
       AWS_REGION: ${AWS_REGION}
-      LAVINMQ_HOST: ${LAVINMQ_HOST}
 
     ports:
       - 3000:3000

package.json

@@ -20,8 +20,8 @@
     },
     "dependencies": {
         "@aws-sdk/client-s3": "^3.936.0",
+        "@aws-sdk/client-transcribe": "^3.948.0",
         "@aws-sdk/s3-request-presigner": "^3.936.0",
-        "@cloudamqp/amqp-client": "^3.4.0",
         "@getbrevo/brevo": "^3.0.1",
         "@prisma/client": "^6.19.0",
         "@quixo3/prisma-session-store": "^3.1.13",

prisma/migrations/20251222131114_add_transcription_job_table/migration.sql

@@ -0,0 +1,23 @@
+-- CreateEnum
+CREATE TYPE "TranscriptionJobStatus" AS ENUM ('PENDING', 'IN_PROGRESS', 'COMPLETED');
+
+-- CreateTable
+CREATE TABLE "TranscriptionJob" (
+    "id" SERIAL NOT NULL,
+    "analysis_entry_id" TEXT NOT NULL,
+    "status" "TranscriptionJobStatus" NOT NULL DEFAULT 'PENDING',
+    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "language_code" TEXT NOT NULL,
+
+    CONSTRAINT "TranscriptionJob_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "TranscriptionJob_id_key" ON "TranscriptionJob"("id");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "TranscriptionJob_analysis_entry_id_key" ON "TranscriptionJob"("analysis_entry_id");
+
+-- AddForeignKey
+ALTER TABLE "TranscriptionJob" ADD CONSTRAINT "TranscriptionJob_analysis_entry_id_fkey" FOREIGN KEY ("analysis_entry_id") REFERENCES "AnalysisEntry"("id") ON DELETE RESTRICT ON UPDATE CASCADE;

prisma/schema.prisma

@@ -44,18 +44,18 @@ model Company {
 }
 
 model ParticipantProfile {
-  id              Int               @id @default(autoincrement())
-  name            String?
-  last_name       String?
-  age             String? // String to allow for prefer not to say option
-  gender          Genders?
-  invited_by      String?
-  country         String?
-  devices         String[]          @default([])
-  created_at      DateTime          @default(now())
-  updated_at      DateTime          @default(now()) @updatedAt
-  User            User              @relation(fields: [user_id], references: [id])
-  user_id         String            @unique
+  id            Int             @id @default(autoincrement())
+  name          String?
+  last_name     String?
+  age           String? // String to allow for prefer not to say option
+  gender        Genders?
+  invited_by    String?
+  country       String?
+  devices       String[]        @default([])
+  created_at    DateTime        @default(now())
+  updated_at    DateTime        @default(now()) @updatedAt
+  User          User            @relation(fields: [user_id], references: [id])
+  user_id       String          @unique
   AnalysisEntry AnalysisEntry[]
 }
 
@@ -85,11 +85,11 @@ model Session {
 }
 
 model Subscription {
-  id         String   @id @unique @default(uuid())
-  company_id String   @unique
-  Company    Company  @relation(fields: [company_id], references: [id])
-  created_at DateTime @default(now())
-  updated_at DateTime @default(now()) @updatedAt
+  id         String    @id @unique @default(uuid())
+  company_id String    @unique
+  Company    Company   @relation(fields: [company_id], references: [id])
+  created_at DateTime  @default(now())
+  updated_at DateTime  @default(now()) @updatedAt
   expires_at DateTime?
 }
 
@@ -105,30 +105,47 @@ model Analysis {
 
   tasks                      Json
   url                        String
-  status                     AnalysisStatus    @default(draft)
-  created_at                 DateTime          @default(now())
-  updated_at                 DateTime          @default(now()) @updatedAt
+  status                     AnalysisStatus  @default(draft)
+  created_at                 DateTime        @default(now())
+  updated_at                 DateTime        @default(now()) @updatedAt
   max_number_of_participants Int
-  AnalysisEntry            AnalysisEntry[]
+  AnalysisEntry              AnalysisEntry[]
 
   @@index([owner_company_id]) // Not a unique field, but requires an index
 }
 
 model AnalysisEntry {
-  id                 String                        @id @default(uuid())
-  Analysis           Analysis                      @relation(fields: [analysis_id], references: [id])
-  analysis_id        String
-  ParticipantProfile ParticipantProfile?           @relation(fields: [user_id], references: [user_id])
-  user_id            String?
-  status             AnalysisEntryCompletionStatus @default(in_progress)
-  created_at         DateTime                      @default(now())
-  updated_at         DateTime                      @default(now()) @updatedAt
+  id                     String                        @id @default(uuid())
+  Analysis               Analysis                      @relation(fields: [analysis_id], references: [id])
+  analysis_id            String
+  ParticipantProfile     ParticipantProfile?           @relation(fields: [user_id], references: [user_id])
+  user_id                String?
+  status                 AnalysisEntryCompletionStatus @default(in_progress)
+  created_at             DateTime                      @default(now())
+  updated_at             DateTime                      @default(now()) @updatedAt
   transcription_segments Json?
-  full_transcript String?
+  full_transcript        String?
+  transcriptionJob       TranscriptionJob?
 
   @@index([analysis_id, status])
 }
 
+model TranscriptionJob {
+  id                Int                    @id @unique @default(autoincrement())
+  AnalysisEntry     AnalysisEntry          @relation(fields: [analysis_entry_id], references: [id])
+  analysis_entry_id String                 @unique
+  status            TranscriptionJobStatus @default(PENDING)
+  created_at        DateTime               @default(now())
+  updated_at        DateTime               @default(now()) @updatedAt
+  language_code     String
+}
+
+enum TranscriptionJobStatus {
+  PENDING
+  IN_PROGRESS
+  COMPLETED
+}
+
 enum Genders {
   male
   female

server/app.js

@@ -12,7 +12,6 @@ import { slowLimiter } from './middlewares/express-slow-down.js';
 import { startCronJobs } from './cron/jobsContainer.js';
 import { globalErrorHandler } from './middlewares/globalErrorHandler.js';
 import { webhookRouter } from './webhooks/webhooksRouter.js';
-import { connectToMessageBroker } from './config/messageBroker/LavinMQ.js';
 
 const app = express();
 const server = createServer(app);
@@ -47,10 +46,9 @@ app.use(globalErrorHandler);
 //* Start the server
 server.listen(process.env.PORT, () => {
 // eslint-disable-next-line no-console
-  console.log(`Server running at http://localhost:${process.env.PORT}/`);
+  console.log('Server running');
 });
 
-connectToMessageBroker();
 startCronJobs();
 
 const gracefulShutdown = () => {

server/controllers/analysisEntryController.js

@@ -1,7 +1,6 @@
-import { logError } from '../config/loggerFunctions.js';
-import { publishToTranscriptionRequestedQueue } from '../config/messageBroker/LavinMQ.js';
 import { generateS3GetPresignedUrl, generateS3PutPresignedUrl } from '../integrations/aws/s3.js';
-import { createAnalysisEntryInDb, getAnalysisEntryDetailsById, updateAnalysisEntryInDb } from '../models/analysisEntryModel.js';
+import { createAnalysisEntryInDb, getAnalysisEntryDetailsById, markAnalysisEntryAsSubmitted } from '../models/analysisEntryModel.js';
+import { processTranscriptionRequest } from '../services/analysisService.js';
 
 export const createAnalysisEntry = async (req, res) => {
   const { analysisId } = req.body;
@@ -16,28 +15,19 @@ export const createAnalysisEntry = async (req, res) => {
 };
 
 export const updateAnalysisEntry = async (req, res) => {
-  const { analysisEntryId, analysisEntryStatus, analysisId } = req.body;
+  const { analysisEntryId } = req.body;
 
-  await updateAnalysisEntryInDb(analysisEntryId, analysisEntryStatus);
+  const updatedAnalysisEntry = await markAnalysisEntryAsSubmitted(analysisEntryId);
 
-  try {
-    const message = {
-      analysisEntryId: analysisEntryId,
-      analysisId: analysisId,
-      timestamp: new Date().toISOString(),
-      mediaType: 'video',
-      languageCode: 'es-ES',
-    };
+  const transcriptionRequest = {
+    analysisEntryId: analysisEntryId,
+    analysisId: updatedAnalysisEntry.analysis_id,
+    languageCode: 'es-ES',
+  };
 
-    const stringifiedMessage = JSON.stringify(message);
-
-    await publishToTranscriptionRequestedQueue(stringifiedMessage);
-  } catch (error) {
-    logError(`Error sending transcription request analysisEntry ${analysisEntryId}`, error);
-  }
+  processTranscriptionRequest(transcriptionRequest); // Fire-and-forget
 
   return res.status(200).json({
-    success: true,
     message: 'Analysis entry updated successfully',
   });
 };

server/cron/getCompletedTranscriptionJobsScheduler.js

@@ -0,0 +1,12 @@
+import { CronJob } from 'cron';
+import { handleCompletedVideoTranscriptionJobs } from '../services/analysisService.js';
+import { logError, logInfo } from '../config/loggerFunctions.js';
+
+export const getCompletedTranscriptionJobsScheduler = new CronJob('15 * * * *', async () => {
+  try {
+    logInfo('Checking transcription job status');
+    await handleCompletedVideoTranscriptionJobs();
+  } catch (error) {
+    logError('Error checking transcription job status', error);
+  }
+});