feat: Enhancements to MessageQueue Module with Decorators

[magrinj]

Overview

This PR introduces significant enhancements to the MessageQueue module by integrating @Processor, @Process, and @InjectMessageQueue decorators. These changes streamline the process of defining and managing queue processors and job handlers, and also allow for request-scoped handlers, improving compatibility with services that rely on scoped providers like TwentyORM repositories.

Key Features

1. Decorator-based Job Handling: Use @Processor and @Process decorators to define job handlers declaratively.
2. Request Scope Support: Job handlers can be scoped per request, enhancing integration with request-scoped services.

Usage

Defining Processors and Job Handlers

The @Processor decorator is used to define a class that processes jobs for a specific queue. The @Process decorator is applied to methods within this class to define specific job handlers.

Example 1: Specific Job Handlers

import { Processor, Process, InjectMessageQueue } from 'src/engine/integrations/message-queue';

@Processor('taskQueue')
export class TaskProcessor {

  @Process('taskA')
  async handleTaskA(job: { id: string, data: any }) {
    console.log(`Handling task A with data:`, job.data);
    // Logic for task A
  }

  @Process('taskB')
  async handleTaskB(job: { id: string, data: any }) {
    console.log(`Handling task B with data:`, job.data);
    // Logic for task B
  }
}

In the example above, TaskProcessor is responsible for processing jobs in the taskQueue. The handleTaskA method will only be called for jobs with the name taskA, while handleTaskB will be called for taskB jobs.

Example 2: General Job Handler

import { Processor, Process, InjectMessageQueue } from 'src/engine/integrations/message-queue';

@Processor('generalQueue')
export class GeneralProcessor {

  @Process()
  async handleAnyJob(job: { id: string, name: string, data: any }) {
    console.log(`Handling job ${job.name} with data:`, job.data);
    // Logic for any job
  }
}

In this example, GeneralProcessor handles all jobs in the generalQueue, regardless of the job name. The handleAnyJob method will be invoked for every job added to the generalQueue.

Adding Jobs to a Queue

You can use the @InjectMessageQueue decorator to inject a queue into a service and add jobs to it.

Example:

import { Injectable } from '@nestjs/common';
import { InjectMessageQueue, MessageQueue } from 'src/engine/integrations/message-queue';

@Injectable()
export class TaskService {
  constructor(
    @InjectMessageQueue('taskQueue') private readonly taskQueue: MessageQueue,
  ) {}

  async addTaskA(data: any) {
    await this.taskQueue.add('taskA', data);
  }

  async addTaskB(data: any) {
    await this.taskQueue.add('taskB', data);
  }
}

In this example, TaskService adds jobs to the taskQueue. The addTaskA and addTaskB methods add jobs named taskA and taskB, respectively, to the queue.

Using Scoped Job Handlers

To utilize request-scoped job handlers, specify the scope in the @Processor decorator. This is particularly useful for services that use scoped repositories like those in TwentyORM.

Example:

import { Processor, Process, InjectMessageQueue, Scope } from 'src/engine/integrations/message-queue';

@Processor({ name: 'scopedQueue', scope: Scope.REQUEST })
export class ScopedTaskProcessor {

  @Process('scopedTask')
  async handleScopedTask(job: { id: string, data: any }) {
    console.log(`Handling scoped task with data:`, job.data);
    // Logic for scoped task, which might use request-scoped services
  }
}

Here, the ScopedTaskProcessor is associated with scopedQueue and operates with request scope. This setup is essential when the job handler relies on services that need to be instantiated per request, such as scoped repositories.

Migration Notes

• Decorators: Refactor job handlers to use @Processor and @Process decorators.
• Request Scope: Utilize the scope option in @Processor if your job handlers depend on request-scoped services.

Fix #5628

[greptile-apps]

PR Summary

• Introduced @Processor and @Process decorators for declarative job handling
• Added @InjectMessageQueue decorator for injecting message queues
• Enhanced MessageQueueModule to support request-scoped job handlers
• Updated various job and listener files to use new decorators
• Integrated MessageQueueModule with conditional registration based on environment variables

[Weiko]

For future improvement, maybe @process() could automatically use the class name directly if not overwritten? I feel like most of the time we will have 1 class per job and always use the handle method

[magrinj]

I'll have to think a bit about this, I'm not 100% sure it's the good way to go

[Weiko]

This seems to be an option for a worker, right? Do we expect to see worker options inside messageQueue options? I guess this is correct as long as we keep with having 1 worker per queue but it could be confusing 🤔. Also this is on a process level but we could have multiple process for the same queue, right? Wdyt?

[greptile-apps]

PR Summary

(updates since last review)

• Introduced @Processor, @Process, and @InjectMessageQueue decorators for enhanced job handling in MessageQueue module.
• Added new environment variable AUTH_TOKEN for GraphQL code generation in twenty-chrome-extension.
• Updated various components and hooks in twenty-front to align with new onboarding and email sync features.
• Refactored multiple job handlers in twenty-server to use new decorator-based approach for better readability and request scope support.
• Introduced new KeyValuePair entity and related modules for improved user state management.

[greptile-apps]

PR Summary

(updates since last review)

• Introduced decorator-based job handling in MessageQueue module
• Added request scope support for job handlers
• Enhanced exception handling in message-queue.explorer.ts
• Updated app.module.ts for conditional MessageQueueModule registration
• Modified TypeORM integration and removed TwentyORMModule from core-engine.module.ts

[Weiko]

LGTM, great job @magrinj!! 👏

cc @charlesBochet