feat: refactor all jobs

Author: magrinj

packages/twenty-server/src/app.module.ts

@@ -22,9 +22,10 @@ import { GraphQLConfigModule } from 'src/engine/api/graphql/graphql-config/graph
 import { GraphQLConfigService } from 'src/engine/api/graphql/graphql-config/graphql-config.service';
 import { WorkspaceCacheVersionModule } from 'src/engine/metadata-modules/workspace-cache-version/workspace-cache-version.module';
 import { GraphQLHydrateRequestFromTokenMiddleware } from 'src/engine/middlewares/graphql-hydrate-request-from-token.middleware';
+import { MessageQueueModule } from 'src/engine/integrations/message-queue/message-queue.module';
 
-import { CoreEngineModule } from './engine/core-modules/core-engine.module';
 import { IntegrationsModule } from './engine/integrations/integrations.module';
+import { CoreEngineModule } from './engine/core-modules/core-engine.module';
 
 @Module({
   imports: [
@@ -74,6 +75,13 @@ export class AppModule {
       );
     }
 
+    // Messaque Queue explorer only for sync driver
+    // Maybe we don't need to conditionaly register the explorer, because we're creating a jobs module
+    // that will expose classes that are only used in the queue worker
+    if (process.env.MESSAGE_QUEUE_TYPE === 'sync') {
+      modules.push(MessageQueueModule.registerExplorer());
+    }
+
     return modules;
   }
 

packages/twenty-server/src/database/commands/data-seed-demo-workspace/jobs/data-seed-demo-workspace.job.ts

@@ -1,15 +1,15 @@
-import { Injectable } from '@nestjs/common';
-
-import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
-
 import { DataSeedDemoWorkspaceService } from 'src/database/commands/data-seed-demo-workspace/services/data-seed-demo-workspace.service';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 
-@Injectable()
-export class DataSeedDemoWorkspaceJob implements MessageQueueJob<undefined> {
+@Processor(MessageQueue.cronQueue)
+export class DataSeedDemoWorkspaceJob {
   constructor(
     private readonly dataSeedDemoWorkspaceService: DataSeedDemoWorkspaceService,
   ) {}
 
+  @Process()
   async handle(): Promise<void> {
     await this.dataSeedDemoWorkspaceService.seedDemo();
   }

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/jobs/call-webhook-jobs.job.ts

@@ -1,6 +1,5 @@
-import { Injectable, Logger } from '@nestjs/common';
+import { Logger } from '@nestjs/common';
 
-import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
 import { ObjectMetadataInterface } from 'src/engine/metadata-modules/field-metadata/interfaces/object-metadata.interface';
 
 import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
@@ -12,6 +11,8 @@ import {
   CallWebhookJobData,
 } from 'src/engine/api/graphql/workspace-query-runner/jobs/call-webhook.job';
 import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 
 export enum CallWebhookJobsJobOperation {
   create = 'create',
@@ -26,10 +27,8 @@ export type CallWebhookJobsJobData = {
   operation: CallWebhookJobsJobOperation;
 };
 
-@Injectable()
-export class CallWebhookJobsJob
-  implements MessageQueueJob<CallWebhookJobsJobData>
-{
+@Processor(MessageQueue.webhookQueue)
+export class CallWebhookJobsJob {
   private readonly logger = new Logger(CallWebhookJobsJob.name);
 
   constructor(
@@ -39,6 +38,7 @@ export class CallWebhookJobsJob
     private readonly messageQueueService: MessageQueueService,
   ) {}
 
+  @Process()
   async handle(data: CallWebhookJobsJobData): Promise<void> {
     const dataSourceMetadata =
       await this.dataSourceService.getLastDataSourceMetadataFromWorkspaceIdOrFail(

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/jobs/call-webhook.job.ts

@@ -1,7 +1,9 @@
-import { Injectable, Logger } from '@nestjs/common';
+import { Logger } from '@nestjs/common';
 import { HttpService } from '@nestjs/axios';
 
-import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 
 export type CallWebhookJobData = {
   targetUrl: string;
@@ -13,12 +15,13 @@ export type CallWebhookJobData = {
   record: any;
 };
 
-@Injectable()
-export class CallWebhookJob implements MessageQueueJob<CallWebhookJobData> {
+@Processor(MessageQueue.webhookQueue)
+export class CallWebhookJob {
   private readonly logger = new Logger(CallWebhookJob.name);
 
   constructor(private readonly httpService: HttpService) {}
 
+  @Process()
   async handle(data: CallWebhookJobData): Promise<void> {
     try {
       await this.httpService.axiosRef.post(data.targetUrl, data);

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/jobs/record-position-backfill.job.ts

@@ -1,22 +1,20 @@
-import { Injectable } from '@nestjs/common';
-
-import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
-
 import { RecordPositionBackfillService } from 'src/engine/api/graphql/workspace-query-runner/services/record-position-backfill-service';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 
 export type RecordPositionBackfillJobData = {
   workspaceId: string;
   dryRun: boolean;
 };
 
-@Injectable()
-export class RecordPositionBackfillJob
-  implements MessageQueueJob<RecordPositionBackfillJobData>
-{
+@Processor(MessageQueue.recordPositionBackfillQueue)
+export class RecordPositionBackfillJob {
   constructor(
     private readonly recordPositionBackfillService: RecordPositionBackfillService,
   ) {}
 
+  @Process()
   async handle(data: RecordPositionBackfillJobData): Promise<void> {
     this.recordPositionBackfillService.backfill(data.workspaceId, data.dryRun);
   }

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/jobs/workspace-query-runner-job.module.ts

@@ -15,19 +15,6 @@ import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/works
     RecordPositionBackfillModule,
     HttpModule,
   ],
-  providers: [
-    {
-      provide: CallWebhookJobsJob.name,
-      useClass: CallWebhookJobsJob,
-    },
-    {
-      provide: CallWebhookJob.name,
-      useClass: CallWebhookJob,
-    },
-    {
-      provide: RecordPositionBackfillJob.name,
-      useClass: RecordPositionBackfillJob,
-    },
-  ],
+  providers: [CallWebhookJobsJob, CallWebhookJob, RecordPositionBackfillJob],
 })
 export class WorkspaceQueryRunnerJobModule {}

packages/twenty-server/src/engine/core-modules/billing/jobs/update-subscription.job.ts

@@ -1,15 +1,15 @@
-import { Injectable, Logger } from '@nestjs/common';
-
-import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
+import { Logger } from '@nestjs/common';
 
 import { BillingService } from 'src/engine/core-modules/billing/billing.service';
 import { UserWorkspaceService } from 'src/engine/core-modules/user-workspace/user-workspace.service';
 import { StripeService } from 'src/engine/core-modules/billing/stripe/stripe.service';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 export type UpdateSubscriptionJobData = { workspaceId: string };
-@Injectable()
-export class UpdateSubscriptionJob
-  implements MessageQueueJob<UpdateSubscriptionJobData>
-{
+
+@Processor(MessageQueue.billingQueue)
+export class UpdateSubscriptionJob {
   protected readonly logger = new Logger(UpdateSubscriptionJob.name);
 
   constructor(
@@ -18,6 +18,7 @@ export class UpdateSubscriptionJob
     private readonly stripeService: StripeService,
   ) {}
 
+  @Process()
   async handle(data: UpdateSubscriptionJobData): Promise<void> {
     const workspaceMembersCount =
       await this.userWorkspaceService.getWorkspaceMemberCount(data.workspaceId);

packages/twenty-server/src/engine/core-modules/core-engine.module.ts

@@ -11,7 +11,6 @@ import { TimelineCalendarEventModule } from 'src/engine/core-modules/calendar/ti
 import { BillingModule } from 'src/engine/core-modules/billing/billing.module';
 import { HealthModule } from 'src/engine/core-modules/health/health.module';
 import { TwentyORMModule } from 'src/engine/twenty-orm/twenty-orm.module';
-import { ShareContextTestModule } from 'src/engine/core-modules/share-context-test/share-context-test.module';
 
 import { AnalyticsModule } from './analytics/analytics.module';
 import { FileModule } from './file/file.module';
@@ -35,7 +34,6 @@ import { ClientConfigModule } from './client-config/client-config.module';
     TimelineCalendarEventModule,
     UserModule,
     WorkspaceModule,
-    ShareContextTestModule,
   ],
   exports: [
     AnalyticsModule,

packages/twenty-server/src/engine/core-modules/share-context-test/share-context-test.module.ts

@@ -1,19 +1,18 @@
-import { Injectable } from '@nestjs/common';
-
-import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
-
 import { WorkspaceService } from 'src/engine/core-modules/workspace/services/workspace.service';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 
 export type HandleWorkspaceMemberDeletedJobData = {
   workspaceId: string;
   userId: string;
 };
-@Injectable()
-export class HandleWorkspaceMemberDeletedJob
-  implements MessageQueueJob<HandleWorkspaceMemberDeletedJobData>
-{
+
+@Processor(MessageQueue.workspaceQueue)
+export class HandleWorkspaceMemberDeletedJob {
   constructor(private readonly workspaceService: WorkspaceService) {}
 
+  @Process()
   async handle(data: HandleWorkspaceMemberDeletedJobData): Promise<void> {
     const { workspaceId, userId } = data;
 

packages/twenty-server/src/engine/core-modules/share-context-test/shared.service.ts

@@ -1,15 +1,15 @@
-import { Injectable } from '@nestjs/common';
-
 import { SendMailOptions } from 'nodemailer';
 
-import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
-
 import { EmailSenderService } from 'src/engine/integrations/email/email-sender.service';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 
-@Injectable()
-export class EmailSenderJob implements MessageQueueJob<SendMailOptions> {
+@Processor(MessageQueue.emailQueue)
+export class EmailSenderJob {
   constructor(private readonly emailSenderService: EmailSenderService) {}
 
+  @Process()
   async handle(data: SendMailOptions): Promise<void> {
     await this.emailSenderService.send(data);
   }

packages/twenty-server/src/engine/core-modules/share-context-test/test.job.ts

@@ -30,7 +30,7 @@ import { MessageQueueModule } from './message-queue/message-queue.module';
       useFactory: loggerModuleFactory,
       inject: [EnvironmentService],
     }),
-    MessageQueueModule.forRoot({
+    MessageQueueModule.registerAsync({
       useFactory: messageQueueModuleFactory,
       inject: [EnvironmentService],
     }),