diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts index 2df94eb643..72372a7d3e 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts @@ -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, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/commands/calendar-trigger-event-list-fetch.command.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/commands/calendar-trigger-event-list-fetch.command.ts new file mode 100644 index 0000000000..2d65c26538 --- /dev/null +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/commands/calendar-trigger-event-list-fetch.command.ts @@ -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 { + 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( + workspaceId, + 'calendarChannel', + ); + + const whereCondition: Record = { + 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( + 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 ', + 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; + } +} diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts index e3b9905593..c870978ed3 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts @@ -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