Reorganise calendar module (#6089)

Refactor Calendar into functional sub modules
<img width="437" alt="image"
src="https://github.com/twentyhq/twenty/assets/12035771/d9de3285-a226-4fe8-b3ef-2d8a21def2a5">

---------

Co-authored-by: bosiraphael <raphael.bosi@gmail.com>

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-events-import.cron.command.ts

@@ -0,0 +1,31 @@
+import { Command, CommandRunner } from 'nest-commander';
+
+import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
+import { CalendarEventsImportCronJob } from 'src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job';
+
+const CALENDAR_EVENTS_IMPORT_CRON_PATTERN = '*/5 * * * *';
+
+@Command({
+  name: 'cron:calendar:calendar-events-import',
+  description: 'Starts a cron job to import calendar events',
+})
+export class CalendarEventsImportCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>(
+      CalendarEventsImportCronJob.name,
+      undefined,
+      {
+        repeat: { pattern: CALENDAR_EVENTS_IMPORT_CRON_PATTERN },
+      },
+    );
+  }
+}

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts

@@ -0,0 +1,72 @@
+import { InjectRepository } from '@nestjs/typeorm';
+import { Scope } from '@nestjs/common';
+
+import { Repository, In } from 'typeorm';
+
+import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-source.entity';
+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';
+import {
+  CalendarEventsImportJobData,
+  CalendarEventsImportJob,
+} from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job';
+import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
+import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
+import { InjectWorkspaceRepository } from 'src/engine/twenty-orm/decorators/inject-workspace-repository.decorator';
+import { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
+import { BillingService } from 'src/engine/core-modules/billing/billing.service';
+import { CalendarChannelWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity';
+
+@Processor({
+  queueName: MessageQueue.cronQueue,
+  scope: Scope.REQUEST,
+})
+export class CalendarEventsImportCronJob {
+  constructor(
+    @InjectRepository(DataSourceEntity, 'metadata')
+    private readonly dataSourceRepository: Repository<DataSourceEntity>,
+    @InjectWorkspaceRepository(CalendarChannelWorkspaceEntity)
+    private readonly calendarChannelRepository: WorkspaceRepository<CalendarChannelWorkspaceEntity>,
+    @InjectMessageQueue(MessageQueue.calendarQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly billingService: BillingService,
+  ) {}
+
+  @Process(CalendarEventsImportCronJob.name)
+  async handle(): Promise<void> {
+    const workspaceIds =
+      await this.billingService.getActiveSubscriptionWorkspaceIds();
+
+    const dataSources = await this.dataSourceRepository.find({
+      where: {
+        workspaceId: In(workspaceIds),
+      },
+    });
+
+    const workspaceIdsWithDataSources = new Set(
+      dataSources.map((dataSource) => dataSource.workspaceId),
+    );
+
+    for (const workspaceId of workspaceIdsWithDataSources) {
+      const calendarChannels = await this.calendarChannelRepository.find({});
+
+      for (const calendarChannel of calendarChannels) {
+        if (!calendarChannel?.isSyncEnabled) {
+          continue;
+        }
+
+        await this.messageQueueService.add<CalendarEventsImportJobData>(
+          CalendarEventsImportJob.name,
+          {
+            workspaceId,
+            connectedAccountId: calendarChannel.connectedAccount.id,
+          },
+          {
+            retryLimit: 2,
+          },
+        );
+      }
+    }
+  }
+}