chore: change the quota sync to the queue

src/app.module.ts

@@ -5,7 +5,6 @@ import { CacheModule } from "@nestjs/cache-manager";
 import { MiddlewareConsumer, Module, NestModule } from "@nestjs/common";
 import { ConfigModule } from "@nestjs/config";
 import { JwtModule } from "@nestjs/jwt";
-import { ScheduleModule } from "@nestjs/schedule";
 import { ThrottlerModule } from "@nestjs/throttler";
 
 import { bullMqConfig } from "./configs/bullmq.config";
@@ -30,7 +29,6 @@ import { UtilsModule } from "./modules/utils/utils.module";
   imports: [
     ConfigModule.forRoot({ isGlobal: true, cache: true }),
     JwtModule.registerAsync(jwtConfig()),
-    ScheduleModule.forRoot(),
     BullModule.forRootAsync(bullMqConfig()),
     CacheModule.registerAsync(cacheConfig()),
     HttpModule.registerAsync(httpConfig()),

src/modules/quota-sync/constants/index.ts

@@ -0,0 +1,11 @@
+export const QUOTA_SYNC_QUEUE = Object.freeze({
+  QUEUE_NAME: "quota-sync",
+  QUEUE_PREFIX: "quota",
+
+  // Job names
+  SYNC_USER_QUOTA_JOB: "sync.user.quota",
+  SYNC_BUSINESS_QUOTA_JOB: "sync.business.quota",
+  SYNC_ALL_USERS_QUOTA_JOB: "sync.all.users.quota",
+  SYNC_ALL_BUSINESSES_QUOTA_JOB: "sync.all.businesses.quota",
+  SYNC_COMPREHENSIVE_QUOTA_JOB: "sync.comprehensive.quota",
+});

src/modules/quota-sync/interfaces/quota-sync-job.interface.ts

@@ -0,0 +1,16 @@
+export interface ISyncUserQuotaJob {
+  userId: string;
+  businessId: string;
+  wildduckUserId: string;
+}
+
+export interface ISyncBusinessQuotaJob {
+  businessId: string;
+  businessName: string;
+}
+
+export type ISyncAllUsersQuotaJob = Record<string, never>;
+
+export type ISyncAllBusinessesQuotaJob = Record<string, never>;
+
+export type ISyncComprehensiveQuotaJob = Record<string, never>;

src/modules/quota-sync/queue/quota-sync.processor.ts

@@ -0,0 +1,109 @@
+// import { EntityManager } from "@mikro-orm/core";
+import { Processor } from "@nestjs/bullmq";
+import { Logger } from "@nestjs/common";
+import { Job } from "bullmq";
+
+import { WorkerProcessor } from "../../../common/queues/worker.processor";
+import { QUOTA_SYNC_QUEUE } from "../constants";
+import {
+  ISyncAllBusinessesQuotaJob,
+  ISyncAllUsersQuotaJob,
+  ISyncBusinessQuotaJob,
+  ISyncComprehensiveQuotaJob,
+  ISyncUserQuotaJob,
+} from "../interfaces/quota-sync-job.interface";
+import { QuotaSyncService } from "../services/quota-sync.service";
+
+@Processor(QUOTA_SYNC_QUEUE.QUEUE_NAME)
+export class QuotaSyncProcessor extends WorkerProcessor {
+  protected readonly logger = new Logger(QuotaSyncProcessor.name);
+
+  constructor(
+    private readonly quotaSyncService: QuotaSyncService,
+    // private readonly em: EntityManager,
+  ) {
+    super();
+  }
+
+  async process(job: Job): Promise<void> {
+    const startTime = Date.now();
+
+    try {
+      await this.handleJob(job);
+      this.logJobSuccess(job, startTime);
+    } catch (error) {
+      this.logJobFailure(job, error, startTime);
+      throw error;
+    }
+  }
+
+  private async handleJob(job: Job): Promise<void> {
+    switch (job.name) {
+      case QUOTA_SYNC_QUEUE.SYNC_USER_QUOTA_JOB:
+        await this.handleSyncUserQuota(job);
+        break;
+      case QUOTA_SYNC_QUEUE.SYNC_BUSINESS_QUOTA_JOB:
+        await this.handleSyncBusinessQuota(job);
+        break;
+      case QUOTA_SYNC_QUEUE.SYNC_ALL_USERS_QUOTA_JOB:
+        await this.handleSyncAllUsersQuota(job);
+        break;
+      case QUOTA_SYNC_QUEUE.SYNC_ALL_BUSINESSES_QUOTA_JOB:
+        await this.handleSyncAllBusinessesQuota(job);
+        break;
+      case QUOTA_SYNC_QUEUE.SYNC_COMPREHENSIVE_QUOTA_JOB:
+        await this.handleSyncComprehensiveQuota(job);
+        break;
+      default:
+        throw new Error(`Unsupported job type: ${job.name}`);
+    }
+  }
+
+  private async handleSyncUserQuota(job: Job<ISyncUserQuotaJob>): Promise<void> {
+    const { userId } = job.data;
+    this.logger.log(`Processing user quota sync for user: ${userId}`);
+
+    await this.quotaSyncService.syncUserQuotaById(userId);
+    this.logger.log(`Successfully synced quota for user: ${userId}`);
+  }
+
+  private async handleSyncBusinessQuota(job: Job<ISyncBusinessQuotaJob>): Promise<void> {
+    const { businessId, businessName } = job.data;
+    this.logger.log(`Processing business quota sync for business: ${businessName} (${businessId})`);
+
+    await this.quotaSyncService.syncBusinessQuotaById(businessId);
+    this.logger.log(`Successfully synced quota for business: ${businessName} (${businessId})`);
+  }
+
+  private async handleSyncAllUsersQuota(_job: Job<ISyncAllUsersQuotaJob>): Promise<void> {
+    this.logger.log("Processing quota sync for all users");
+
+    await this.quotaSyncService.syncAllUsersQuota();
+    this.logger.log("Successfully synced quota for all users");
+  }
+
+  private async handleSyncAllBusinessesQuota(_job: Job<ISyncAllBusinessesQuotaJob>): Promise<void> {
+    this.logger.log("Processing quota sync for all businesses");
+
+    await this.quotaSyncService.syncAllBusinessQuotas();
+    this.logger.log("Successfully synced quota for all businesses");
+  }
+
+  private async handleSyncComprehensiveQuota(_job: Job<ISyncComprehensiveQuotaJob>): Promise<void> {
+    this.logger.log("Processing comprehensive quota sync");
+
+    await this.quotaSyncService.syncAllQuotas();
+    this.logger.log("Successfully completed comprehensive quota sync");
+  }
+
+  private logJobSuccess(job: Job, startTime: number): void {
+    const duration = Date.now() - startTime;
+    this.logger.log(`Job ${job.id} (${job.name}) completed successfully in ${duration}ms`);
+  }
+
+  private logJobFailure(job: Job, error: unknown, startTime: number): void {
+    const duration = Date.now() - startTime;
+    const errorMessage = error instanceof Error ? error.message : "Unknown error";
+    this.logger.error(`Job ${job.id} (${job.name}) failed after ${duration}ms: ${errorMessage}`);
+  }
+}

src/modules/quota-sync/quota-sync.module.ts

@@ -1,11 +1,24 @@
+import { BullModule } from "@nestjs/bullmq";
 import { Module } from "@nestjs/common";
 
-import { MailServerModule } from "../mail-server/mail-server.module";
+import { QUOTA_SYNC_QUEUE } from "./constants";
+import { QuotaSyncProcessor } from "./queue/quota-sync.processor";
 import { QuotaSyncService } from "./services/quota-sync.service";
+import { MailServerModule } from "../mail-server/mail-server.module";
 
 @Module({
-  imports: [MailServerModule],
-  providers: [QuotaSyncService],
+  imports: [
+    MailServerModule,
+    BullModule.registerQueue({
+      name: QUOTA_SYNC_QUEUE.QUEUE_NAME,
+      prefix: QUOTA_SYNC_QUEUE.QUEUE_PREFIX,
+      defaultJobOptions: {
+        removeOnComplete: true,
+        removeOnFail: true,
+      },
+    }),
+  ],
+  providers: [QuotaSyncService, QuotaSyncProcessor],
   exports: [QuotaSyncService],
 })
 export class QuotaSyncModule {}

src/modules/quota-sync/services/quota-sync.service.ts

@@ -1,11 +1,13 @@
 import { EntityManager } from "@mikro-orm/core";
+import { InjectQueue } from "@nestjs/bullmq";
 import { Injectable, Logger } from "@nestjs/common";
-import { Cron, CronExpression } from "@nestjs/schedule";
+import { Queue } from "bullmq";
 import { firstValueFrom } from "rxjs";
 
 import { Business } from "../../businesses/entities/business.entity";
 import { WildDuckUsersService } from "../../mail-server/services/wildduck-users.service";
 import { User } from "../../users/entities/user.entity";
+import { QUOTA_SYNC_QUEUE } from "../constants";
 import {
   BusinessQuotaInfoDto,
   BusinessQuotaStatisticsDto,
@@ -14,20 +16,127 @@ import {
   QuotaSyncResultDto,
   UserQuotaStatisticsDto,
 } from "../DTO/quota-statistics.dto";
+import { ISyncBusinessQuotaJob, ISyncUserQuotaJob } from "../interfaces/quota-sync-job.interface";
 
 @Injectable()
 export class QuotaSyncService {
   private readonly logger = new Logger(QuotaSyncService.name);
 
   constructor(
+    @InjectQueue(QUOTA_SYNC_QUEUE.QUEUE_NAME) private readonly quotaSyncQueue: Queue,
     private readonly wildDuckUsersService: WildDuckUsersService,
     private readonly em: EntityManager,
   ) {}
 
   /**
-   * Sync quota usage every 2 hours
+   * Add a job to sync quota usage for all users
+   */
+  async scheduleAllUsersQuotaSync(): Promise<void> {
+    await this.quotaSyncQueue.add(
+      QUOTA_SYNC_QUEUE.SYNC_ALL_USERS_QUOTA_JOB,
+      {},
+      {
+        removeOnComplete: true,
+        removeOnFail: false,
+        attempts: 3,
+        backoff: {
+          type: "exponential",
+          delay: 2000,
+        },
+      },
+    );
+    this.logger.log("Scheduled all users quota sync job");
+  }
+
+  /**
+   * Add a job to sync business quota usage for all businesses
+   */
+  async scheduleAllBusinessQuotasSync(): Promise<void> {
+    await this.quotaSyncQueue.add(
+      QUOTA_SYNC_QUEUE.SYNC_ALL_BUSINESSES_QUOTA_JOB,
+      {},
+      {
+        removeOnComplete: true,
+        removeOnFail: false,
+        attempts: 3,
+        backoff: {
+          type: "exponential",
+          delay: 2000,
+        },
+      },
+    );
+    this.logger.log("Scheduled all businesses quota sync job");
+  }
+
+  /**
+   * Add a job to sync quota for a specific user
+   */
+  async scheduleUserQuotaSync(userId: string, businessId: string, wildduckUserId: string): Promise<void> {
+    const jobData: ISyncUserQuotaJob = {
+      userId,
+      businessId,
+      wildduckUserId,
+    };
+
+    await this.quotaSyncQueue.add(QUOTA_SYNC_QUEUE.SYNC_USER_QUOTA_JOB, jobData, {
+      removeOnComplete: true,
+      removeOnFail: false,
+      attempts: 3,
+      backoff: {
+        type: "exponential",
+        delay: 2000,
+      },
+    });
+    this.logger.log(`Scheduled user quota sync job for user: ${userId}`);
+  }
+
+  /**
+   * Add a job to sync quota for a specific business
+   */
+  async scheduleBusinessQuotaSync(businessId: string, businessName: string): Promise<void> {
+    const jobData: ISyncBusinessQuotaJob = {
+      businessId,
+      businessName,
+    };
+
+    await this.quotaSyncQueue.add(QUOTA_SYNC_QUEUE.SYNC_BUSINESS_QUOTA_JOB, jobData, {
+      removeOnComplete: true,
+      removeOnFail: false,
+      attempts: 3,
+      backoff: {
+        type: "exponential",
+        delay: 2000,
+      },
+    });
+    this.logger.log(`Scheduled business quota sync job for business: ${businessName}`);
+  }
+
+  /**
+   * Add a job for comprehensive quota sync
+   */
+  async scheduleComprehensiveQuotaSync(): Promise<void> {
+    await this.quotaSyncQueue.add(
+      QUOTA_SYNC_QUEUE.SYNC_COMPREHENSIVE_QUOTA_JOB,
+      {},
+      {
+        removeOnComplete: true,
+        removeOnFail: false,
+        attempts: 3,
+        backoff: {
+          type: "exponential",
+          delay: 2000,
+        },
+      },
+    );
+    this.logger.log("Scheduled comprehensive quota sync job");
+  }
+
+  // The following methods remain the same but without cron decorators
+  // They are now called by the queue processor
+
+  /**
+   * Sync quota usage for all users (called by processor)
    */
-  @Cron(CronExpression.EVERY_2_HOURS)
   async syncAllUsersQuota(): Promise<void> {
     const em = this.em.fork();
     this.logger.log("Starting quota synchronization for all users");
@@ -60,13 +169,13 @@ export class QuotaSyncService {
       this.logger.log(`Quota sync completed: ${successCount} success, ${errorCount} errors`);
     } catch (error) {
       this.logger.error("Failed to sync quota for users", `${error instanceof Error ? error.message : "Unknown error"}`);
+      throw error;
     }
   }
 
   /**
-   * Sync business quota usage every 4 hours
+   * Sync business quota usage for all businesses (called by processor)
    */
-  @Cron(CronExpression.EVERY_4_HOURS)
   async syncAllBusinessQuotas(): Promise<void> {
     const em = this.em.fork();
     this.logger.log("Starting business quota synchronization");
@@ -97,6 +206,7 @@ export class QuotaSyncService {
       this.logger.log(`Business quota sync completed: ${successCount} success, ${errorCount} errors`);
     } catch (error) {
       this.logger.error("Failed to sync quota for businesses", `${error instanceof Error ? error.message : "Unknown error"}`);
+      throw error;
     }
   }
 

src/modules/templates/DTO/index.ts

@@ -1,4 +0,0 @@
-export { CreateTemplateDto } from "./create-template.dto";
-export { UpdateTemplateDto } from "./update-template.dto";
-export { TemplateQueryDto } from "./template-query.dto";
-export { TemplateParamDto } from "./template-param.dto";

src/modules/templates/DTO/update-template.dto.ts

@@ -1,44 +1,5 @@
-import { ApiPropertyOptional } from "@nestjs/swagger";
-import { Type } from "class-transformer";
-import { IsOptional, IsString, MaxLength, ValidateNested } from "class-validator";
+import { PartialType } from "@nestjs/swagger";
 
-import { TemplateMessage } from "../../../common/enums/message.enum";
-import { PersonalityDataType } from "../interfaces/structure.interface";
+import { CreateTemplateDto } from "./create-template.dto";
 
-export class UpdateTemplateDto {
-  @IsOptional()
-  @IsString({ message: TemplateMessage.TEMPLATE_NAME_STRING })
-  @MaxLength(255, { message: TemplateMessage.TEMPLATE_NAME_MAX_LENGTH })
-  @ApiPropertyOptional({ description: "Template name", example: "Updated Corporate Email Template", maxLength: 255 })
-  name?: string;
-
-  @IsOptional()
-  @ValidateNested({ message: TemplateMessage.TEMPLATE_STRUCTURE_VALID })
-  @Type(() => Object)
-  @ApiPropertyOptional({
-    description: "Template structure containing header, content, and footer sections",
-    example: {
-      header: {
-        columnsCount: 1,
-        items: [],
-      },
-      content: {
-        columnsCount: 2,
-        items: [],
-      },
-      footer: {
-        columnsCount: 1,
-        items: [],
-      },
-    },
-  })
-  structure?: PersonalityDataType;
-
-  @IsOptional()
-  @IsString({ message: TemplateMessage.TEMPLATE_RAW_HTML_STRING })
-  @ApiPropertyOptional({
-    description: "Raw HTML content of the template",
-    example: "<html><body><h1>Updated Template Content</h1></body></html>",
-  })
-  rawHtml?: string;
-}
+export class UpdateTemplateDto extends PartialType(CreateTemplateDto) {}

src/modules/users/services/users.service.ts

@@ -10,6 +10,7 @@ import { DomainAutomationService } from "../../domains/services/domain-automatio
 import { DomainsService } from "../../domains/services/domains.service";
 import { UpdateUserDto } from "../../mail-server/DTO";
 import { MailServerService } from "../../mail-server/services/mail-server.service";
+import { QuotaSyncService } from "../../quota-sync/services/quota-sync.service";
 import { Template } from "../../templates/entities/template.entity";
 import { PasswordService } from "../../utils/services/password.service";
 import { CreateEmailUserDto } from "../DTO/create-email-user.dto";
@@ -31,6 +32,7 @@ export class UsersService {
     private readonly mailServerService: MailServerService,
     private readonly domainsService: DomainsService,
     private readonly domainAutomationService: DomainAutomationService,
+    private readonly quotaSyncService: QuotaSyncService,
   ) {}
 
   async getUserEmailIdFromId(userId: string) {
@@ -169,6 +171,11 @@ export class UsersService {
 
       await this.em.flush();
 
+      //sync quota
+      await this.quotaSyncService.scheduleUserQuotaSync(user.id, businessId, mailServerUser.id);
+      //sync business quota
+      await this.quotaSyncService.scheduleBusinessQuotaSync(businessId, business.name);
+
       this.logger.log(`Email user created: ${emailAddress} for business ${businessId}`);
 
       await this.domainAutomationService.createDefaultMailboxes(mailServerUser.id);
@@ -221,6 +228,11 @@ export class UsersService {
 
       await this.em.persistAndFlush(user);
 
+      //sync quota
+      await this.quotaSyncService.scheduleUserQuotaSync(user.id, businessId, user.wildduckUserId);
+      //sync business quota
+      await this.quotaSyncService.scheduleBusinessQuotaSync(businessId, user.business.name);
+
       this.logger.log(`Email user deleted: ${user.emailAddress} for business ${businessId}`);
 
       return { message: UserMessage.EMAIL_USER_DELETED_SUCCESSFULLY };
@@ -316,6 +328,14 @@ export class UsersService {
 
       await this.em.flush();
 
+      //sync quota
+      if (user.wildduckUserId) {
+        await this.quotaSyncService.scheduleUserQuotaSync(user.id, businessId, user.wildduckUserId);
+      }
+
+      //sync business quota
+      await this.quotaSyncService.scheduleBusinessQuotaSync(businessId, user.business.name);
+
       this.logger.log(`Email user updated: ${user.emailAddress} for business ${businessId}`);
 
       return {
@@ -372,6 +392,14 @@ export class UsersService {
 
       await this.em.persistAndFlush([user, user.business]);
 
+      //sync quota
+      if (user.wildduckUserId) {
+        await this.quotaSyncService.scheduleUserQuotaSync(user.id, businessId, user.wildduckUserId);
+      }
+
+      //sync business quota
+      await this.quotaSyncService.scheduleBusinessQuotaSync(businessId, user.business.name);
+
       this.logger.log(`Email user quota updated: ${user.emailAddress} to ${newQuota} bytes`);
 
       return { message: UserMessage.EMAIL_USER_QUOTA_UPDATED_SUCCESSFULLY, user };

src/modules/users/users.module.ts

@@ -9,10 +9,11 @@ import { UsersController } from "./users.controller";
 import { BusinessesModule } from "../businesses/businesses.module";
 import { DomainsModule } from "../domains/domains.module";
 import { MailServerModule } from "../mail-server/mail-server.module";
+import { QuotaSyncModule } from "../quota-sync/quota-sync.module";
 import { UtilsModule } from "../utils/utils.module";
 
 @Module({
-  imports: [MikroOrmModule.forFeature([User, RefreshToken, Role]), UtilsModule, MailServerModule, DomainsModule, BusinessesModule],
+  imports: [MikroOrmModule.forFeature([User, RefreshToken, Role]), UtilsModule, MailServerModule, DomainsModule, BusinessesModule, QuotaSyncModule],
   controllers: [UsersController],
   providers: [UsersService],
   exports: [UsersService],