fix calendar save events transaction (#16763)

2026-01-06 23:26:03 +05:30
parent ceba0972cd
commit 971ddc0bb4
3 changed files with 245 additions and 107 deletions
@@ -10,6 +10,7 @@ import { ObjectMetadataRepositoryModule } from 'src/engine/object-metadata-repos
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
 import { CalendarEventCleanerModule } from 'src/modules/calendar/calendar-event-cleaner/calendar-event-cleaner.module';
+import { CalendarTriggerEventListFetchCommand } from 'src/modules/calendar/calendar-event-import-manager/commands/calendar-trigger-event-list-fetch.command';
 import { CalendarEventListFetchCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-event-list-fetch.cron.command';
 import { CalendarEventsImportCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-import.cron.command';
 import { CalendarOngoingStaleCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-ongoing-stale.cron.command';
@@ -73,6 +74,7 @@ import { RefreshTokensManagerModule } from 'src/modules/connected-account/refres
    CalendarEventsImportJob,
    CalendarOngoingStaleCronJob,
    CalendarOngoingStaleCronCommand,
+    CalendarTriggerEventListFetchCommand,
    CalendarOngoingStaleJob,
    CalendarRelaunchFailedCalendarChannelsCronJob,
    CalendarRelaunchFailedCalendarChannelsCronCommand,
@@ -0,0 +1,132 @@
+import { Logger } from '@nestjs/common';
+
+import { Command, CommandRunner, Option } from 'nest-commander';
+
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
+import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
+import {
+  CalendarEventListFetchJob,
+  type CalendarEventListFetchJobData,
+} from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job';
+import {
+  CalendarChannelSyncStage,
+  type CalendarChannelWorkspaceEntity,
+} from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity';
+
+type CalendarTriggerEventListFetchCommandOptions = {
+  workspaceId: string;
+  calendarChannelId?: string;
+};
+
+@Command({
+  name: 'calendar:trigger-event-list-fetch',
+  description:
+    'Trigger calendar event list fetch immediately without waiting for cron',
+})
+export class CalendarTriggerEventListFetchCommand extends CommandRunner {
+  private readonly logger = new Logger(
+    CalendarTriggerEventListFetchCommand.name,
+  );
+
+  constructor(
+    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    @InjectMessageQueue(MessageQueue.calendarQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(
+    _passedParam: string[],
+    options: CalendarTriggerEventListFetchCommandOptions,
+  ): Promise<void> {
+    const { workspaceId, calendarChannelId } = options;
+
+    this.logger.log(
+      `Triggering calendar event list fetch for workspace ${workspaceId}${calendarChannelId ? ` and channel ${calendarChannelId}` : ' (all pending channels)'}`,
+    );
+
+    const authContext = buildSystemAuthContext(workspaceId);
+
+    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
+      authContext,
+      async () => {
+        const calendarChannelRepository =
+          await this.globalWorkspaceOrmManager.getRepository<CalendarChannelWorkspaceEntity>(
+            workspaceId,
+            'calendarChannel',
+          );
+
+        const whereCondition: Record<string, unknown> = {
+          isSyncEnabled: true,
+          syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
+        };
+
+        if (calendarChannelId) {
+          whereCondition.id = calendarChannelId;
+        }
+
+        const calendarChannels =
+          await calendarChannelRepository.find(whereCondition);
+
+        if (calendarChannels.length === 0) {
+          this.logger.warn(
+            'No calendar channels found with CALENDAR_EVENT_LIST_FETCH_PENDING status',
+          );
+
+          return;
+        }
+
+        this.logger.log(
+          `Found ${calendarChannels.length} calendar channel(s) to process`,
+        );
+
+        for (const calendarChannel of calendarChannels) {
+          await calendarChannelRepository.update(calendarChannel.id, {
+            syncStage:
+              CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED,
+            syncStageStartedAt: new Date().toISOString(),
+          });
+
+          await this.messageQueueService.add<CalendarEventListFetchJobData>(
+            CalendarEventListFetchJob.name,
+            {
+              calendarChannelId: calendarChannel.id,
+              workspaceId,
+            },
+          );
+
+          this.logger.log(
+            `Triggered fetch for calendar channel ${calendarChannel.id}`,
+          );
+        }
+
+        this.logger.log(
+          `Successfully triggered ${calendarChannels.length} calendar event list fetch job(s)`,
+        );
+      },
+    );
+  }
+
+  @Option({
+    flags: '-w, --workspace-id <workspace_id>',
+    description: 'Workspace ID',
+    required: true,
+  })
+  parseWorkspaceId(value: string): string {
+    return value;
+  }
+
+  @Option({
+    flags: '-c, --calendar-channel-id [calendar_channel_id]',
+    description:
+      'Calendar Channel ID (optional - if not provided, triggers for all pending channels)',
+    required: false,
+  })
+  parseCalendarChannelId(value: string): string {
+    return value;
+  }
+}
@@ -1,6 +1,7 @@
 import { Injectable } from '@nestjs/common';

 import { Any } from 'typeorm';
+import { v4 as uuid } from 'uuid';

 import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
 import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
@@ -48,42 +49,108 @@ export class CalendarSaveEventsService {
            'calendarChannelEventAssociation',
          );

-        const existingCalendarEvents = await calendarEventRepository.find({
-          where: {
-            iCalUid: Any(
-              fetchedCalendarEvents.map((event) => event.iCalUid as string),
-            ),
-          },
-        });
-
-        const fetchedCalendarEventsWithDBEvents: FetchedCalendarEventWithDBEvent[] =
-          fetchedCalendarEvents.map(
-            (event): FetchedCalendarEventWithDBEvent => {
-              const existingEventWithSameiCalUid = existingCalendarEvents.find(
-                (existingEvent) => existingEvent.iCalUid === event.iCalUid,
-              );
-
-              return {
-                fetchedCalendarEvent: event,
-                existingCalendarEvent: existingEventWithSameiCalUid ?? null,
-                newlyCreatedCalendarEvent: null,
-              };
-            },
-          );
-
        const workspaceDataSource =
          await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource();

        await workspaceDataSource.transaction(
          async (transactionManager: WorkspaceEntityManager) => {
-            const savedCalendarEvents = await calendarEventRepository.save(
-              fetchedCalendarEventsWithDBEvents
+            const existingCalendarEvents = await calendarEventRepository.find(
+              {
+                where: {
+                  iCalUid: Any(
+                    fetchedCalendarEvents.map(
+                      (event) => event.iCalUid as string,
+                    ),
+                  ),
+                },
+              },
+              transactionManager,
+            );
+
+            const fetchedCalendarEventsWithDBEvents: FetchedCalendarEventWithDBEvent[] =
+              fetchedCalendarEvents.map(
+                (event): FetchedCalendarEventWithDBEvent => {
+                  const existingEventWithSameiCalUid =
+                    existingCalendarEvents.find(
+                      (existingEvent) =>
+                        existingEvent.iCalUid === event.iCalUid,
+                    );
+
+                  return {
+                    fetchedCalendarEvent: event,
+                    existingCalendarEvent: existingEventWithSameiCalUid ?? null,
+                    newlyCreatedCalendarEvent: null,
+                  };
+                },
+              );
+
+            const newCalendarEventsToInsert = fetchedCalendarEventsWithDBEvents
+              .filter(
+                ({ existingCalendarEvent }) => existingCalendarEvent === null,
+              )
+              .map(({ fetchedCalendarEvent }) => ({
+                id: uuid(),
+                iCalUid: fetchedCalendarEvent.iCalUid,
+                title: fetchedCalendarEvent.title,
+                description: fetchedCalendarEvent.description,
+                startsAt: fetchedCalendarEvent.startsAt,
+                endsAt: fetchedCalendarEvent.endsAt,
+                location: fetchedCalendarEvent.location,
+                isFullDay: fetchedCalendarEvent.isFullDay,
+                isCanceled: fetchedCalendarEvent.isCanceled,
+                conferenceSolution: fetchedCalendarEvent.conferenceSolution,
+                conferenceLink: {
+                  primaryLinkLabel: fetchedCalendarEvent.conferenceLinkLabel,
+                  primaryLinkUrl: fetchedCalendarEvent.conferenceLinkUrl,
+                  secondaryLinks: [],
+                },
+                externalCreatedAt: fetchedCalendarEvent.externalCreatedAt,
+                externalUpdatedAt: fetchedCalendarEvent.externalUpdatedAt,
+              }));
+
+            if (newCalendarEventsToInsert.length > 0) {
+              await calendarEventRepository.insert(
+                newCalendarEventsToInsert,
+                transactionManager,
+              );
+            }
+
+            const fetchedCalendarEventsWithDBEventsEnrichedWithSavedEvents: FetchedCalendarEventWithDBEvent[] =
+              fetchedCalendarEventsWithDBEvents.map(
+                ({ fetchedCalendarEvent, existingCalendarEvent }) => {
+                  const savedCalendarEvent = newCalendarEventsToInsert.find(
+                    (inserted) =>
+                      inserted.iCalUid === fetchedCalendarEvent.iCalUid,
+                  );
+
+                  return {
+                    fetchedCalendarEvent,
+                    existingCalendarEvent: existingCalendarEvent,
+                    newlyCreatedCalendarEvent: savedCalendarEvent
+                      ? ({
+                          id: savedCalendarEvent.id,
+                          iCalUid: savedCalendarEvent.iCalUid,
+                        } as CalendarEventWorkspaceEntity)
+                      : null,
+                  };
+                },
+              );
+
+            const existingEventsToUpdate =
+              fetchedCalendarEventsWithDBEventsEnrichedWithSavedEvents
                .filter(
-                  ({ existingCalendarEvent }) => existingCalendarEvent === null,
+                  ({ existingCalendarEvent }) => existingCalendarEvent !== null,
                )
-                .map(
-                  ({ fetchedCalendarEvent }) =>
-                    ({
+                .map(({ fetchedCalendarEvent, existingCalendarEvent }) => {
+                  if (!existingCalendarEvent) {
+                    throw new Error(
+                      `Existing calendar event with iCalUid ${fetchedCalendarEvent.iCalUid} not found - should never happen`,
+                    );
+                  }
+
+                  return {
+                    criteria: existingCalendarEvent.id,
+                    partialEntity: {
                      iCalUid: fetchedCalendarEvent.iCalUid,
                      title: fetchedCalendarEvent.title,
                      description: fetchedCalendarEvent.description,
@@ -102,80 +169,16 @@ export class CalendarSaveEventsService {
                      },
                      externalCreatedAt: fetchedCalendarEvent.externalCreatedAt,
                      externalUpdatedAt: fetchedCalendarEvent.externalUpdatedAt,
-                    }) satisfies Omit<
-                      CalendarEventWorkspaceEntity,
-                      | 'id'
-                      | 'calendarChannelEventAssociations'
-                      | 'calendarEventParticipants'
-                      | 'createdAt'
-                      | 'updatedAt'
-                      | 'deletedAt'
-                    >,
-                ),
-              {},
-              transactionManager,
-            );
-
-            const fetchedCalendarEventsWithDBEventsEnrichedWithSavedEvents: FetchedCalendarEventWithDBEvent[] =
-              fetchedCalendarEventsWithDBEvents.map(
-                ({ fetchedCalendarEvent, existingCalendarEvent }) => {
-                  const savedCalendarEvent = savedCalendarEvents.find(
-                    (savedCalendarEvent) =>
-                      savedCalendarEvent.iCalUid ===
-                      fetchedCalendarEvent.iCalUid,
-                  );
-
-                  return {
-                    fetchedCalendarEvent,
-                    existingCalendarEvent: existingCalendarEvent,
-                    newlyCreatedCalendarEvent: savedCalendarEvent ?? null,
-                  };
-                },
-              );
-
-            await calendarEventRepository.save(
-              fetchedCalendarEventsWithDBEventsEnrichedWithSavedEvents
-                .filter(
-                  ({ existingCalendarEvent }) => existingCalendarEvent !== null,
-                )
-                .map(({ fetchedCalendarEvent, existingCalendarEvent }) => {
-                  if (!existingCalendarEvent) {
-                    throw new Error(
-                      `Existing calendar event with iCalUid ${fetchedCalendarEvent.iCalUid} not found - should never happen`,
-                    );
-                  }
-
-                  return {
-                    id: existingCalendarEvent.id,
-                    iCalUid: fetchedCalendarEvent.iCalUid,
-                    title: fetchedCalendarEvent.title,
-                    description: fetchedCalendarEvent.description,
-                    startsAt: fetchedCalendarEvent.startsAt,
-                    endsAt: fetchedCalendarEvent.endsAt,
-                    location: fetchedCalendarEvent.location,
-                    isFullDay: fetchedCalendarEvent.isFullDay,
-                    isCanceled: fetchedCalendarEvent.isCanceled,
-                    conferenceSolution: fetchedCalendarEvent.conferenceSolution,
-                    conferenceLink: {
-                      primaryLinkLabel:
-                        fetchedCalendarEvent.conferenceLinkLabel,
-                      primaryLinkUrl: fetchedCalendarEvent.conferenceLinkUrl,
-                      secondaryLinks: [],
                    },
-                    externalCreatedAt: fetchedCalendarEvent.externalCreatedAt,
-                    externalUpdatedAt: fetchedCalendarEvent.externalUpdatedAt,
-                  } satisfies Omit<
-                    CalendarEventWorkspaceEntity,
-                    | 'calendarChannelEventAssociations'
-                    | 'calendarEventParticipants'
-                    | 'createdAt'
-                    | 'updatedAt'
-                    | 'deletedAt'
-                  >;
-                }),
-              {},
-              transactionManager,
-            );
+                  };
+                });
+
+            if (existingEventsToUpdate.length > 0) {
+              await calendarEventRepository.updateMany(
+                existingEventsToUpdate,
+                transactionManager,
+              );
+            }

            const calendarChannelEventAssociationsToSave: Pick<
              CalendarChannelEventAssociationWorkspaceEntity,
@@ -208,11 +211,12 @@ export class CalendarSaveEventsService {
              },
            );

-            await calendarChannelEventAssociationRepository.save(
-              calendarChannelEventAssociationsToSave,
-              {},
-              transactionManager,
-            );
+            if (calendarChannelEventAssociationsToSave.length > 0) {
+              await calendarChannelEventAssociationRepository.insert(
+                calendarChannelEventAssociationsToSave,
+                transactionManager,
+              );
+            }

            const participantsToCreate =
              fetchedCalendarEventsWithDBEventsEnrichedWithSavedEvents