REFACTOR AWS transcription to local transcription

.env.example

@@ -23,9 +23,8 @@ AWS_SECRET_ACCESS_KEY=
 AWS_ACCESS_KEY_ID=
 S3_REGION=
 S3_BUCKET=
-S3_ENDPOINT=
-
-TRANSCRIPTION_ENABLED=
+S3_EXTERNAL_ENDPOINT=
+S3_INTERNAL_ENDPOINT=
 
 MINIO_ROOT_USER=
 MINIO_ROOT_PASSWORD=

compose.yaml

@@ -1,4 +1,18 @@
 name: uxcaptain
+
+networks:
+  uxcaptain-network:
+    name: uxcaptain-network
+    driver: bridge
+
+volumes:
+  uxcaptain-database:
+    name: uxcaptain-database
+
+  whisper-transcription-models:
+    name: whisper-transcription-models
+
+
 services:
   # server:
   #   container_name: server
@@ -18,13 +32,13 @@ services:
   #     STRIPE_API_KEY: ${STRIPE_API_KEY}
   #     STRIPE_WEBHOOK_SECRET: ${STRIPE_WEBHOOK_SECRET}
   #     PORT: ${PORT}
-  #     AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
-  #     AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
   #     S3_BUCKET: dev-analysis-entry-storage
   #     S3_REGION: ${S3_REGION}
-  #     S3_ENDPOINT: ${S3_ENDPOINT}
+  #     S3_EXTERNAL_ENDPOINT: ${S3_EXTERNAL_ENDPOINT}
+#       S3_INTERNAL_ENDPOINT: ${S3_EXTERNAL_ENDPOINT} # Has to be internal to allow in-server communication
   #     MINIO_ROOT_USER=${MINIO_ROOT_USER}
   #     MINIO_ROOT_PASSWORD=${MINIO_ROOT_PASSWORD}
+  #     TRANSCRIPTION_ENDPOINT= ${TRANSCRIPTION_ENDPOINT}
 
   #   ports:
   #     - 3000:3000
@@ -40,6 +54,15 @@ services:
           # condition: service_started
   #   networks:
   #     - uxcaptain-network
+
+  #   deploy:
+  #       resources:
+  #         limits:
+  #           cpus: '1.0'
+  #           memory: 512M
+  #         reservations:
+  #           cpus: '0.5'
+  #           memory: 256M
 
   minio:
     image: minio/minio:latest
@@ -74,11 +97,30 @@ services:
     networks:
       - uxcaptain-network 
 
-networks:
-  uxcaptain-network:
-    name: uxcaptain-network
-    driver: bridge
+  faster-whisper-transcribe:
+      image: onerahmet/openai-whisper-asr-webservice:latest
+      container_name: faster-whisper-asr
+      ports:
+        - 9007:9000
+      volumes:
+        - whisper-transcription-models:/root/.cache  # Model persistence (~2-5GB)
+      environment:
+        - ASR_MODEL=medium  # tiny,base,small,medium,large https://ahmetoner.com/whisper-asr-webservice/environmental-variables/#configuring-the-model
+        - ASR_DEVICE=cpu # https://ahmetoner.com/whisper-asr-webservice/environmental-variables/#configuring-device-and-quantization
+        - ASR_ENGINE=faster_whisper # openai_whisper, faster_whisper, whisperx -- https://ahmetoner.com/whisper-asr-webservice/environmental-variables/#whisperx
+        - MODEL_IDLE_TIMEOUT=0  # in Seconds - Keep model loaded - https://ahmetoner.com/whisper-asr-webservice/environmental-variables/#configuring-the-model-unloading-timeout
+        - ASR_QUANTIZATION=int8 # https://ahmetoner.com/whisper-asr-webservice/environmental-variables/#configuring-device-and-quantization
+      restart: unless-stopped
+      depends_on:
+        - minio
+      networks:
+        - uxcaptain-network  # Same as MinIO/monolith
+      deploy:
+        resources:
+          limits:
+            cpus: '4.0'
+            memory: 5000M
+
+
 
-volumes:
-  uxcaptain-database:
 

package.json

@@ -20,7 +20,6 @@
     },
     "dependencies": {
         "@aws-sdk/client-s3": "^3.936.0",
-        "@aws-sdk/client-transcribe": "^3.948.0",
         "@aws-sdk/s3-request-presigner": "^3.936.0",
         "@getbrevo/brevo": "^3.0.1",
         "@prisma/client": "^6.19.0",
@@ -35,6 +34,7 @@
         "express-session": "^1.18.2",
         "express-slow-down": "^3.0.1",
         "express-validator": "^7.3.1",
+        "form-data": "^4.0.5",
         "helmet": "^8.1.0",
         "npm": "^11.6.3",
         "passport": "^0.7.0",

server/controllers/analysisController.js

@@ -7,7 +7,7 @@ import {
   from '../models/analysisModel.js';
 
 import { createAnalysisEntryInDb } from '../models/analysisEntryModel.js';
-import { generateS3PutPresignedUrl } from '../integrations/aws/s3.js';
+import { generateS3PutPresignedUrl } from '../integrations/s3-client/s3.js';
 
 export const createAnalysis = async (req, res) => {
   if (req.sanitizedErrors) {

server/controllers/analysisEntryController.js

@@ -1,6 +1,7 @@
-import { generateS3GetPresignedUrl } from '../integrations/aws/s3.js';
+import { logError, logInfo } from '../config/loggerFunctions.js';
+import { generateS3GetPresignedUrl } from '../integrations/s3-client/s3.js';
 import { createAnalysisEntryInDb, getAnalysisEntryDetailsById, markAnalysisEntryAsSubmitted } from '../models/analysisEntryModel.js';
-import { processTranscriptionRequest } from '../services/analysisService.js';
+import { insertTranscriptionJobInDb } from '../models/transcriptionModel.js';
 
 export const createAnalysisEntry = async (req, res) => {
   const { analysisId } = req.body;
@@ -19,14 +20,17 @@ export const updateAnalysisEntry = async (req, res) => {
 
   const updatedAnalysisEntry = await markAnalysisEntryAsSubmitted(analysisEntryId);
 
-  const transcriptionRequest = {
+  const transcriptionJob = {
     analysisEntryId: analysisEntryId,
     analysisId: updatedAnalysisEntry.analysis_id,
-    languageCode: 'es-ES',
+    languageCode: 'es',
   };
 
-  if (process.env.TRANSCRIPTION_ENABLED === 'true') {
-    processTranscriptionRequest(transcriptionRequest); // Fire-and-forget
+  try {
+    await insertTranscriptionJobInDb(transcriptionJob);
+    logInfo(`Transcription job for ${transcriptionJob.analysisEntryId} stored in DB`, transcriptionJob);
+  } catch (error) {
+    logError(`error inserting ${transcriptionJob.analysisEntryId} analysisEntry's transcription request`, error);
   }
 
   return res.status(200).json({

server/controllers/authController.js

@@ -172,30 +172,24 @@ export const checkSession = async (req, res) => {
 export const loginLocal = async (req, res, next) => {
   if (req.sanitizedErrors) {
     return res.status(422).json({
-      success: false,
       message: 'Analysis could not be created due to validation errors',
       errors: req.sanitizedErrors,
     });
   }
 
   return passport.authenticate('local', (err, user /* , info */) => {
     if (err) {
-      return res.status(500).json({
-        success: false,
-        message: 'An error occurred during login',
-      });
+      return next(err);
     }
 
     if (!user) { //* Will trigger if user does not exist
       return res.status(401).json({
-        success: false,
         message: 'The combination of email and password is incorrect',
       });
     }
 
     if (user.role === 'participant') {
       return res.status(403).json({
-        success: false,
         message: 'Participant login is disabled',
       });
     }
@@ -215,7 +209,6 @@ export const loginLocal = async (req, res, next) => {
       updateUserLastLoginDate(user.id); // Fire-and-forget function
 
       return res.status(200).json({
-        success: true,
         message: 'Login successful',
         user: {
           id: user.id,

server/controllers/transcriptionController.js

@@ -0,0 +1,45 @@
+import { logError, logInfo } from '../config/loggerFunctions.js';
+import { transcribeRecording } from '../integrations/whisper-asr-webservice/transcribe.js';
+import {
+  getPendingTranscriptionJobsFromDb,
+  storeNormalizedTranscriptionInDb,
+  updateStatusSingleTranscriptionJobInDb,
+} from '../models/transcriptionModel.js';
+import { cleanUpTranscriptSegments } from '../utils/transcription/transcriptionNormalizer.js';
+
+export const processPendingTranscriptionJobs = async () => {
+  const pendingTranscriptionJobs = await getPendingTranscriptionJobsFromDb();
+
+  if (pendingTranscriptionJobs.length === 0) {
+    logInfo('No pending transcription jobs found');
+    return;
+  }
+
+  for (const transcriptionJob of pendingTranscriptionJobs) {
+    logInfo(`Processing transcription job for analysis entry ID: ${transcriptionJob.analysis_entry_id}`);
+
+    const { analysis_entry_id: analysisEntryId } = transcriptionJob;
+
+    try {
+      // Mark job as IN_PROGRESS before making async call to prevent re-queuing
+      await updateStatusSingleTranscriptionJobInDb(analysisEntryId, 'IN_PROGRESS');
+      logInfo(`Marked transcription job ${analysisEntryId} as IN_PROGRESS`);
+
+      const transcriptionJobResult = await transcribeRecording(transcriptionJob);
+
+      const { segments, text: fullText } = transcriptionJobResult;
+
+      const cleanedUpSegments = await cleanUpTranscriptSegments(segments);
+
+      await storeNormalizedTranscriptionInDb(analysisEntryId, fullText, cleanedUpSegments);
+
+      await updateStatusSingleTranscriptionJobInDb(analysisEntryId, 'COMPLETED');
+
+      logInfo(`Transcription job ${analysisEntryId} completed successfully`);
+    } catch (error) {
+      // Mark job back as PENDING to allow retry
+      await updateStatusSingleTranscriptionJobInDb(analysisEntryId, 'PENDING');
+      logError(`Error processing transcription job, ${analysisEntryId}`, error);
+    }
+  }
+};

server/cron/getPendingTranscriptionJobScheduler.js

@@ -0,0 +1,11 @@
+import { CronJob } from 'cron';
+import { processPendingTranscriptionJobs } from '../controllers/transcriptionController.js';
+import { logError } from '../config/loggerFunctions.js';
+
+export const getPendingTranscriptionJobScheduler = new CronJob('*/15 * * * *', async () => {
+  try {
+    await processPendingTranscriptionJobs();
+  } catch (error) {
+    logError('Error processing transcription request', error);
+  }
+});

server/cron/jobsContainer.js

@@ -1,20 +1,16 @@
-import { logError } from '../config/loggerFunctions.js';
+import { logError, logInfo } from '../config/loggerFunctions.js';
 import { deletePasswordResetTokensScheduler } from './deletePasswordResetTokensScheduler.js';
-import { getCompletedTranscriptionJobsScheduler } from './getCompletedTranscriptionJobsScheduler.js';
+import { getPendingTranscriptionJobScheduler } from './getPendingTranscriptionJobScheduler.js';
 import { markAnalysisEntriesAsCancelledScheduler } from './markAsCancelledAnalysisEntriesScheduler.js';
 
 export const startCronJobs = () => {
+  logInfo('Starting cron jobs');
   try {
     deletePasswordResetTokensScheduler.start();
 
-    if (process.env.TRANSCRIPTION_ENABLED === true) {
-      getCompletedTranscriptionJobsScheduler.start();
-    }
+    getPendingTranscriptionJobScheduler.start();
 
     markAnalysisEntriesAsCancelledScheduler.start();
-
-
-    console.log('Cron jobs started');
   } catch (error) {
     logError('error on startCronJobs', error);
   }