refactor: add memory use logs, cleanup queue, streams (#44)

Author: xixas

apps/api/src/jobs/consumers/summary/summary.consumer.ts

@@ -1,6 +1,6 @@
-import { Logger } from '@nestjs/common';
-import { Process, Processor } from '@nestjs/bull';
-import { Job } from 'bull';
+import { Logger, OnModuleDestroy } from '@nestjs/common';
+import { InjectQueue, Process, Processor } from '@nestjs/bull';
+import { Job, Queue } from 'bull';
 import { InjectRepository } from '@nestjs/typeorm';
 import { Repository } from 'typeorm';
 import { Summary } from 'src/modules/summary/entities/summary.entity';
@@ -13,14 +13,15 @@ import { SUMMARY_QUEUE } from 'src/modules/summary/queues/summary.queue';
 import { uploadDir } from 'src/main';
 
 @Processor(SUMMARY_QUEUE)
-export class SummaryConsumer {
+export class SummaryConsumer implements OnModuleDestroy {
   private readonly logger = new Logger(SummaryConsumer.name);
 
   constructor(
     @InjectRepository(Summary)
     private readonly summaryRepository: Repository<Summary>,
     private readonly summaryService: SummaryService,
     private readonly meetingSummarizerService: MeetingSummaryService,
+    @InjectQueue(SUMMARY_QUEUE) private readonly summaryQueue: Queue,
   ) {}
 
   @Process()
@@ -49,8 +50,7 @@ export class SummaryConsumer {
       const summaryText = await this.meetingSummarizerService.generateMeetingSummary(fileContent);
       summary.outputText =
         summaryText ||
-        'Summary could not be generated, please check if its a valid teams transcript file';
-      // Update job status to SUCCESS
+        "Summary could not be generated, please check if it's a valid teams transcript file";
       summary.jobStatus = JobStatus.SUCCESS;
       this.logger.log(`Summary generated successfully for job with ID: ${jobId}`);
     } catch (error) {
@@ -62,4 +62,10 @@ export class SummaryConsumer {
       await this.summaryRepository.save(summary);
     }
   }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.summaryQueue) {
+      await this.summaryQueue.close();
+    }
+  }
 }

apps/api/src/jobs/producers/summary/summary.producer.ts

@@ -1,11 +1,11 @@
-import { Injectable, Logger, Optional } from '@nestjs/common';
+import { Injectable, Logger, Optional, OnModuleDestroy } from '@nestjs/common';
 import { InjectQueue } from '@nestjs/bull';
 import { Queue } from 'bull';
 import { Summary } from 'src/modules/summary/entities/summary.entity';
 import { SUMMARY_QUEUE } from 'src/modules/summary/queues/summary.queue';
 
 @Injectable()
-export class SummaryQueueProducer {
+export class SummaryQueueProducer implements OnModuleDestroy {
   private readonly logger = new Logger(SummaryQueueProducer.name);
 
   constructor(@Optional() @InjectQueue(SUMMARY_QUEUE) private readonly summaryQueue: Queue) {}
@@ -27,7 +27,16 @@ export class SummaryQueueProducer {
     this.logger.log(`Adding job with ID ${summary.jobId}`);
 
     if (this.summaryQueue) {
-      await this.summaryQueue.add(summary.jobId);
+      await this.summaryQueue.add(summary.jobId, {
+        removeOnComplete: true,
+        removeOnFail: true,
+      });
+    }
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.summaryQueue) {
+      await this.summaryQueue.close();
     }
   }
 }

apps/api/src/modules/summary/schedule/schedule.service.ts

@@ -1,13 +1,89 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
 import { Cron, CronExpression } from '@nestjs/schedule';
 import { SummaryService } from '../summary.service';
+import { ConfigService } from '@nestjs/config';
+import Redis from 'ioredis';
 
 @Injectable()
-export class ScheduleService {
-  constructor(private readonly summaryService: SummaryService) {}
+export class ScheduleService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(ScheduleService.name);
+  private isProcessing: boolean = false;
+  private redisClient: Redis;
+
+  constructor(
+    private readonly summaryService: SummaryService,
+    private readonly configService: ConfigService,
+  ) {}
+
+  onModuleInit(): void {
+    this.redisClient = new Redis({
+      host: this.configService.get<string>('REDIS_HOST'),
+      port: this.configService.get<number>('REDIS_PORT'),
+    });
+  }
 
   @Cron(CronExpression.EVERY_SECOND)
   async addSummaryToQueue(): Promise<void> {
-    this.summaryService.addSummaryToQueue();
+    if (this.isProcessing) {
+      return;
+    }
+
+    this.isProcessing = true;
+
+    try {
+      this.logMemoryUsage();
+      await this.summaryService.addSummaryToQueue();
+    } catch (error) {
+      this.logger.error('Error in addSummaryToQueue:', error);
+    } finally {
+      this.isProcessing = false;
+    }
+  }
+
+  private logMemoryUsage(): void {
+    // Node.js memory usage
+    const memoryUsage = process.memoryUsage();
+    this.logger.log(
+      `Memory Usage: RSS = ${memoryUsage.rss / 1024 / 1024} MB, Heap Total = ${memoryUsage.heapTotal / 1024 / 1024} MB, Heap Used = ${memoryUsage.heapUsed / 1024 / 1024} MB`,
+    );
+
+    // Redis memory usage
+    this.redisClient.info('memory', (err, res) => {
+      if (err) {
+        this.logger.error('Error fetching Redis memory info:', err);
+      } else {
+        const memoryInfo = this.parseRedisMemoryInfo(res);
+        this.logger.log(
+          `Redis Memory Usage: Used = ${memoryInfo.used_memory_human}, RSS = ${memoryInfo.used_memory_rss_human}, Peak = ${memoryInfo.used_memory_peak_human}, System Memory = ${memoryInfo.total_system_memory_human}, Fragmentation Ratio = ${memoryInfo.mem_fragmentation_ratio}`,
+        );
+      }
+    });
+  }
+
+  // eslint-disable-next-line
+  private parseRedisMemoryInfo(info: string): any {
+    const memoryInfo = {};
+    const lines = info.split('\r\n');
+    const relevantKeys = [
+      'used_memory_human',
+      'used_memory_rss_human',
+      'used_memory_peak_human',
+      'total_system_memory_human',
+      'mem_fragmentation_ratio',
+    ];
+    lines.forEach((line) => {
+      const [key, value] = line.split(':');
+
+      if (relevantKeys.includes(key)) {
+        memoryInfo[key] = value;
+      }
+    });
+    return memoryInfo;
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.redisClient) {
+      await this.redisClient.quit();
+    }
   }
 }

apps/api/src/modules/summary/summarizer/summarizer.service.ts

@@ -32,13 +32,15 @@ export class MeetingSummaryService {
       .split('\n')
       .map((line) => line.trim())
       .filter((line) => line);
-    const timePattern =
+    const timePattern1 =
       /(\d{1,2}:\d{2}:\d{2}(?:\.\d{1,3})?) --> (\d{1,2}:\d{2}:\d{2}(?:\.\d{1,3})?)/; // Matches time format "HH:MM:SS" or "HH:MM:SS.mmm"
+    const timePattern2 =
+      /(\d{1,2}:\d{1,2}:\d{1,2}(?:\.\d{1,3})?) --> (\d{1,2}:\d{1,2}:\d{1,2}(?:\.\d{1,3})?)/; // Matches time format "H:M:S" or "H:M:S.mmm"
     const chunks: Chunk[] = [];
     let i = 0;
 
     while (i < lines.length) {
-      const match = lines[i].match(timePattern);
+      const match = lines[i].match(timePattern1) || lines[i].match(timePattern2);
 
       if (match) {
         const [startTime, endTime] = match.slice(1, 3);

apps/api/src/modules/summary/summary.service.ts

@@ -46,8 +46,10 @@ export class SummaryService extends CoreService<Summary> {
       const fileLocation = join(uploadDir, modifiedFilename);
 
       return new Promise((resolve, reject) => {
+        const writeStream = createWriteStream(fileLocation);
+
         createReadStream()
-          .pipe(createWriteStream(fileLocation))
+          .pipe(writeStream)
           .on('finish', async () => {
             const summary = this.summaryRepository.create({ inputFile: modifiedFilename });
 
@@ -65,9 +67,12 @@ export class SummaryService extends CoreService<Summary> {
               resolve(savedSummary);
             } catch (error) {
               reject(error);
+            } finally {
+              writeStream.close();
             }
           })
           .on('error', (error) => {
+            writeStream.close();
             reject(error);
           });
       });
@@ -98,6 +103,7 @@ export class SummaryService extends CoreService<Summary> {
       },
     });
   }
+
   getPendingSummary(): Promise<Summary[]> {
     this.logger.log('Getting all active pending summaries');
     return this.summaryRepository.find({
@@ -145,5 +151,6 @@ export class SummaryService extends CoreService<Summary> {
     }
 
     this.isProcessingQueue = false;
+    this.logger.log('Finished processing cron');
   }
 }