diff --git a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-hook/workspace-query-hook.module.ts b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-hook/workspace-query-hook.module.ts index ed18dd3a13..77bd3216ae 100644 --- a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-hook/workspace-query-hook.module.ts +++ b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-hook/workspace-query-hook.module.ts @@ -10,7 +10,6 @@ import { CalendarQueryHookModule } from 'src/modules/calendar/common/query-hooks import { ConnectedAccountQueryHookModule } from 'src/modules/connected-account/query-hooks/connected-account-query-hook.module'; import { DashboardQueryHookModule } from 'src/modules/dashboard/query-hooks/dashboard-query-hook.module'; import { MessagingQueryHookModule } from 'src/modules/messaging/common/query-hooks/messaging-query-hook.module'; -import { MessageChannelQueryHookModule } from 'src/modules/messaging/message-channel-manager/query-hooks/message-channel-query-hook.module'; import { WorkspaceMemberQueryHookModule } from 'src/modules/workspace-member/query-hooks/workspace-member-query-hook.module'; @Module({ @@ -20,7 +19,6 @@ import { WorkspaceMemberQueryHookModule } from 'src/modules/workspace-member/que ConnectedAccountQueryHookModule, DashboardQueryHookModule, BlocklistQueryHookModule, - MessageChannelQueryHookModule, WorkspaceMemberQueryHookModule, DiscoveryModule, ], diff --git a/packages/twenty-server/src/engine/core-modules/auth/auth.module.ts b/packages/twenty-server/src/engine/core-modules/auth/auth.module.ts index 1a99cf3638..7bfe55a976 100644 --- a/packages/twenty-server/src/engine/core-modules/auth/auth.module.ts +++ b/packages/twenty-server/src/engine/core-modules/auth/auth.module.ts @@ -22,9 +22,6 @@ import { CreateMessageChannelService } from 'src/engine/core-modules/auth/servic import { GoogleAPIScopesService } from 'src/engine/core-modules/auth/services/google-apis-scopes'; import { GoogleAPIsService } from 'src/engine/core-modules/auth/services/google-apis.service'; import { MicrosoftAPIsService } from 'src/engine/core-modules/auth/services/microsoft-apis.service'; -import { ResetCalendarChannelService } from 'src/engine/core-modules/auth/services/reset-calendar-channel.service'; -import { ResetMessageChannelService } from 'src/engine/core-modules/auth/services/reset-message-channel.service'; -import { ResetMessageFolderService } from 'src/engine/core-modules/auth/services/reset-message-folder.service'; import { ResetPasswordService } from 'src/engine/core-modules/auth/services/reset-password.service'; import { SignInUpService } from 'src/engine/core-modules/auth/services/sign-in-up.service'; import { UpdateConnectedAccountOnReconnectService } from 'src/engine/core-modules/auth/services/update-connected-account-on-reconnect.service'; @@ -62,7 +59,9 @@ import { UserRoleModule } from 'src/engine/metadata-modules/user-role/user-role. import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module'; import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; import { WorkspaceManagerModule } from 'src/engine/workspace-manager/workspace-manager.module'; +import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; import { ConnectedAccountModule } from 'src/modules/connected-account/connected-account.module'; +import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { MessagingFolderSyncManagerModule } from 'src/modules/messaging/message-folder-manager/messaging-folder-sync-manager.module'; import { TwoFactorAuthenticationMethodEntity } from '../two-factor-authentication/entities/two-factor-authentication-method.entity'; @@ -142,9 +141,8 @@ import { JwtAuthStrategy } from './strategies/jwt.auth.strategy'; // So far, it's not possible to have controllers in business modules // which forces us to have these services in the auth module // TODO: Move these calendar, message, and connected account services to the business modules once possible - ResetMessageChannelService, - ResetCalendarChannelService, - ResetMessageFolderService, + MessageChannelSyncStatusService, + CalendarChannelSyncStatusService, CreateMessageChannelService, CreateCalendarChannelService, CreateConnectedAccountService, diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.spec.ts b/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.spec.ts index 8782ee2e3b..e27ec20ee1 100644 --- a/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.spec.ts +++ b/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.spec.ts @@ -8,20 +8,20 @@ import { CreateConnectedAccountService } from 'src/engine/core-modules/auth/serv import { CreateMessageChannelService } from 'src/engine/core-modules/auth/services/create-message-channel.service'; import { GoogleAPIScopesService } from 'src/engine/core-modules/auth/services/google-apis-scopes'; import { GoogleAPIsService } from 'src/engine/core-modules/auth/services/google-apis.service'; -import { ResetCalendarChannelService } from 'src/engine/core-modules/auth/services/reset-calendar-channel.service'; -import { ResetMessageChannelService } from 'src/engine/core-modules/auth/services/reset-message-channel.service'; import { UpdateConnectedAccountOnReconnectService } from 'src/engine/core-modules/auth/services/update-connected-account-on-reconnect.service'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-queue-token.util'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; import { CalendarChannelSyncStage, CalendarChannelVisibility, } from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity'; import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; +import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { MessageChannelVisibility } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; jest.mock('uuid', () => ({ @@ -30,8 +30,8 @@ jest.mock('uuid', () => ({ describe('GoogleAPIsService', () => { let service: GoogleAPIsService; - let resetCalendarChannelService: ResetCalendarChannelService; - let resetMessageChannelService: ResetMessageChannelService; + let calendarChannelSyncStatusService: CalendarChannelSyncStatusService; + let messagingChannelSyncStatusService: MessageChannelSyncStatusService; let createMessageChannelService: CreateMessageChannelService; const mockConnectedAccountRepository = { @@ -109,9 +109,9 @@ describe('GoogleAPIsService', () => { useValue: mockTwentyConfigService, }, { - provide: ResetCalendarChannelService, + provide: CalendarChannelSyncStatusService, useValue: { - resetCalendarChannels: jest.fn(), + resetAndScheduleCalendarEventListFetch: jest.fn(), }, }, { @@ -125,9 +125,9 @@ describe('GoogleAPIsService', () => { }, }, { - provide: ResetMessageChannelService, + provide: MessageChannelSyncStatusService, useValue: { - resetMessageChannels: jest.fn(), + resetAndScheduleMessageListFetch: jest.fn(), }, }, { @@ -172,12 +172,14 @@ describe('GoogleAPIsService', () => { }).compile(); service = module.get(GoogleAPIsService); - resetCalendarChannelService = module.get( - ResetCalendarChannelService, - ); - resetMessageChannelService = module.get( - ResetMessageChannelService, - ); + calendarChannelSyncStatusService = + module.get( + CalendarChannelSyncStatusService, + ); + messagingChannelSyncStatusService = + module.get( + MessageChannelSyncStatusService, + ); createMessageChannelService = module.get( CreateMessageChannelService, ); @@ -232,20 +234,12 @@ describe('GoogleAPIsService', () => { }); expect( - resetCalendarChannelService.resetCalendarChannels, - ).toHaveBeenCalledWith({ - workspaceId: 'workspace-id', - connectedAccountId: 'existing-account-id', - manager: expect.any(Object), - }); + calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch, + ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id'); expect( - resetMessageChannelService.resetMessageChannels, - ).toHaveBeenCalledWith({ - workspaceId: 'workspace-id', - connectedAccountId: 'existing-account-id', - manager: expect.any(Object), - }); + messagingChannelSyncStatusService.resetAndScheduleMessageListFetch, + ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id'); expect( createMessageChannelService.createMessageChannel, diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.ts b/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.ts index 5c8d7e8be8..6928567cab 100644 --- a/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.ts +++ b/packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.ts @@ -11,8 +11,6 @@ import { CreateCalendarChannelService } from 'src/engine/core-modules/auth/servi import { CreateConnectedAccountService } from 'src/engine/core-modules/auth/services/create-connected-account.service'; import { CreateMessageChannelService } from 'src/engine/core-modules/auth/services/create-message-channel.service'; import { GoogleAPIScopesService } from 'src/engine/core-modules/auth/services/google-apis-scopes'; -import { ResetCalendarChannelService } from 'src/engine/core-modules/auth/services/reset-calendar-channel.service'; -import { ResetMessageChannelService } from 'src/engine/core-modules/auth/services/reset-message-channel.service'; import { UpdateConnectedAccountOnReconnectService } from 'src/engine/core-modules/auth/services/update-connected-account-on-reconnect.service'; 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'; @@ -24,6 +22,7 @@ import { CalendarEventListFetchJob, type CalendarEventListFetchJobData, } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job'; +import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; import { CalendarChannelSyncStage, type CalendarChannelVisibility, @@ -31,6 +30,7 @@ import { } from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity'; import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; +import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { MessageChannelSyncStage, type MessageChannelVisibility, @@ -52,9 +52,9 @@ export class GoogleAPIsService { private readonly calendarQueueService: MessageQueueService, private readonly twentyConfigService: TwentyConfigService, private readonly accountsToReconnectService: AccountsToReconnectService, - private readonly resetMessageChannelService: ResetMessageChannelService, - private readonly resetCalendarChannelService: ResetCalendarChannelService, private readonly createMessageChannelService: CreateMessageChannelService, + private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService, + private readonly calendarChannelSyncStatusService: CalendarChannelSyncStatusService, private readonly createCalendarChannelService: CreateCalendarChannelService, private readonly createConnectedAccountService: CreateConnectedAccountService, private readonly updateConnectedAccountOnReconnectService: UpdateConnectedAccountOnReconnectService, @@ -190,17 +190,15 @@ export class GoogleAPIsService { newOrExistingConnectedAccountId, ); - await this.resetMessageChannelService.resetMessageChannels({ + await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch( + [newOrExistingConnectedAccountId], workspaceId, - connectedAccountId: newOrExistingConnectedAccountId, - manager, - }); + ); - await this.resetCalendarChannelService.resetCalendarChannels({ + await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch( + [newOrExistingConnectedAccountId], workspaceId, - connectedAccountId: newOrExistingConnectedAccountId, - manager, - }); + ); } }, ); diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.spec.ts b/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.spec.ts index 7cddccdade..1b00ae89ce 100644 --- a/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.spec.ts +++ b/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.spec.ts @@ -7,21 +7,20 @@ import { CreateCalendarChannelService } from 'src/engine/core-modules/auth/servi import { CreateConnectedAccountService } from 'src/engine/core-modules/auth/services/create-connected-account.service'; import { CreateMessageChannelService } from 'src/engine/core-modules/auth/services/create-message-channel.service'; import { MicrosoftAPIsService } from 'src/engine/core-modules/auth/services/microsoft-apis.service'; -import { ResetCalendarChannelService } from 'src/engine/core-modules/auth/services/reset-calendar-channel.service'; -import { ResetMessageChannelService } from 'src/engine/core-modules/auth/services/reset-message-channel.service'; -import { ResetMessageFolderService } from 'src/engine/core-modules/auth/services/reset-message-folder.service'; import { UpdateConnectedAccountOnReconnectService } from 'src/engine/core-modules/auth/services/update-connected-account-on-reconnect.service'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-queue-token.util'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; import { CalendarChannelSyncStage, CalendarChannelVisibility, } from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity'; import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; +import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { MessageChannelVisibility } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; jest.mock('uuid', () => ({ @@ -30,8 +29,8 @@ jest.mock('uuid', () => ({ describe('MicrosoftAPIsService', () => { let service: MicrosoftAPIsService; - let resetCalendarChannelService: ResetCalendarChannelService; - let resetMessageChannelService: ResetMessageChannelService; + let messagingChannelSyncStatusService: MessageChannelSyncStatusService; + let calendarChannelSyncStatusService: CalendarChannelSyncStatusService; let createMessageChannelService: CreateMessageChannelService; const mockConnectedAccountRepository = { @@ -109,21 +108,15 @@ describe('MicrosoftAPIsService', () => { useValue: mockTwentyConfigService, }, { - provide: ResetCalendarChannelService, + provide: CalendarChannelSyncStatusService, useValue: { - resetCalendarChannels: jest.fn(), + resetAndScheduleCalendarEventListFetch: jest.fn(), }, }, { - provide: ResetMessageChannelService, + provide: MessageChannelSyncStatusService, useValue: { - resetMessageChannels: jest.fn(), - }, - }, - { - provide: ResetMessageFolderService, - useValue: { - resetMessageFolders: jest.fn(), + resetAndScheduleMessageListFetch: jest.fn(), }, }, { @@ -170,12 +163,14 @@ describe('MicrosoftAPIsService', () => { }).compile(); service = module.get(MicrosoftAPIsService); - resetCalendarChannelService = module.get( - ResetCalendarChannelService, - ); - resetMessageChannelService = module.get( - ResetMessageChannelService, - ); + calendarChannelSyncStatusService = + module.get( + CalendarChannelSyncStatusService, + ); + messagingChannelSyncStatusService = + module.get( + MessageChannelSyncStatusService, + ); createMessageChannelService = module.get( CreateMessageChannelService, ); @@ -235,20 +230,12 @@ describe('MicrosoftAPIsService', () => { }); expect( - resetCalendarChannelService.resetCalendarChannels, - ).toHaveBeenCalledWith({ - workspaceId: 'workspace-id', - connectedAccountId: 'existing-account-id', - manager: expect.any(Object), - }); + calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch, + ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id'); expect( - resetMessageChannelService.resetMessageChannels, - ).toHaveBeenCalledWith({ - workspaceId: 'workspace-id', - connectedAccountId: 'existing-account-id', - manager: expect.any(Object), - }); + messagingChannelSyncStatusService.resetAndScheduleMessageListFetch, + ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id'); expect( createMessageChannelService.createMessageChannel, diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.ts b/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.ts index 07cabe6aae..6270c81987 100644 --- a/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.ts +++ b/packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.ts @@ -6,9 +6,6 @@ import { v4 } from 'uuid'; import { CreateCalendarChannelService } from 'src/engine/core-modules/auth/services/create-calendar-channel.service'; import { CreateConnectedAccountService } from 'src/engine/core-modules/auth/services/create-connected-account.service'; import { CreateMessageChannelService } from 'src/engine/core-modules/auth/services/create-message-channel.service'; -import { ResetCalendarChannelService } from 'src/engine/core-modules/auth/services/reset-calendar-channel.service'; -import { ResetMessageChannelService } from 'src/engine/core-modules/auth/services/reset-message-channel.service'; -import { ResetMessageFolderService } from 'src/engine/core-modules/auth/services/reset-message-folder.service'; import { UpdateConnectedAccountOnReconnectService } from 'src/engine/core-modules/auth/services/update-connected-account-on-reconnect.service'; import { getMicrosoftApisOauthScopes } from 'src/engine/core-modules/auth/utils/get-microsoft-apis-oauth-scopes'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; @@ -21,6 +18,7 @@ import { CalendarEventListFetchJob, type CalendarEventListFetchJobData, } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job'; +import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; import { CalendarChannelSyncStage, type CalendarChannelVisibility, @@ -28,6 +26,7 @@ import { } from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity'; import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; +import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { MessageChannelSyncStage, type MessageChannelVisibility, @@ -48,9 +47,8 @@ export class MicrosoftAPIsService { @InjectMessageQueue(MessageQueue.calendarQueue) private readonly calendarQueueService: MessageQueueService, private readonly accountsToReconnectService: AccountsToReconnectService, - private readonly resetMessageChannelService: ResetMessageChannelService, - private readonly resetMessageFolderService: ResetMessageFolderService, - private readonly resetCalendarChannelService: ResetCalendarChannelService, + private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService, + private readonly calendarChannelSyncStatusService: CalendarChannelSyncStatusService, private readonly createMessageChannelService: CreateMessageChannelService, private readonly createCalendarChannelService: CreateCalendarChannelService, private readonly createConnectedAccountService: CreateConnectedAccountService, @@ -175,23 +173,20 @@ export class MicrosoftAPIsService { newOrExistingConnectedAccountId, ); - await this.resetMessageChannelService.resetMessageChannels({ + await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch( + [newOrExistingConnectedAccountId], workspaceId, - connectedAccountId: newOrExistingConnectedAccountId, - manager, - }); + ); - await this.resetMessageFolderService.resetMessageFolders({ + await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch( + [newOrExistingConnectedAccountId], workspaceId, - connectedAccountId: newOrExistingConnectedAccountId, - manager, - }); + ); - await this.resetCalendarChannelService.resetCalendarChannels({ + await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch( + [newOrExistingConnectedAccountId], workspaceId, - connectedAccountId: newOrExistingConnectedAccountId, - manager, - }); + ); } }, ); diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/reset-calendar-channel.service.ts b/packages/twenty-server/src/engine/core-modules/auth/services/reset-calendar-channel.service.ts deleted file mode 100644 index f350cd8584..0000000000 --- a/packages/twenty-server/src/engine/core-modules/auth/services/reset-calendar-channel.service.ts +++ /dev/null @@ -1,48 +0,0 @@ -import { Injectable } from '@nestjs/common'; - -import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; -import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { - CalendarChannelSyncStage, - type CalendarChannelWorkspaceEntity, -} from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity'; - -export type ResetCalendarChannelsInput = { - workspaceId: string; - connectedAccountId: string; - manager: WorkspaceEntityManager; -}; - -@Injectable() -export class ResetCalendarChannelService { - constructor( - private readonly twentyORMGlobalManager: TwentyORMGlobalManager, - ) {} - - async resetCalendarChannels( - input: ResetCalendarChannelsInput, - ): Promise { - const { workspaceId, connectedAccountId, manager } = input; - - const calendarChannelRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - 'calendarChannel', - ); - - await calendarChannelRepository.update( - { - connectedAccountId, - }, - { - syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING, - syncStatus: null, - syncCursor: '', - syncStageStartedAt: null, - }, - manager, - ); - - return; - } -} diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/reset-message-channel.service.ts b/packages/twenty-server/src/engine/core-modules/auth/services/reset-message-channel.service.ts deleted file mode 100644 index 5ba9bcf91f..0000000000 --- a/packages/twenty-server/src/engine/core-modules/auth/services/reset-message-channel.service.ts +++ /dev/null @@ -1,47 +0,0 @@ -import { Injectable } from '@nestjs/common'; - -import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; -import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { - MessageChannelSyncStage, - MessageChannelSyncStatus, - type MessageChannelWorkspaceEntity, -} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; - -export type ResetMessageChannelsInput = { - workspaceId: string; - connectedAccountId: string; - manager: WorkspaceEntityManager; -}; - -@Injectable() -export class ResetMessageChannelService { - constructor( - private readonly twentyORMGlobalManager: TwentyORMGlobalManager, - ) {} - - async resetMessageChannels(input: ResetMessageChannelsInput): Promise { - const { workspaceId, connectedAccountId, manager } = input; - - const messageChannelRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - 'messageChannel', - ); - - await messageChannelRepository.update( - { - connectedAccountId, - }, - { - syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING, - syncStatus: MessageChannelSyncStatus.ONGOING, - syncCursor: '', - syncStageStartedAt: null, - }, - manager, - ); - - return; - } -} diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/reset-message-folder.service.ts b/packages/twenty-server/src/engine/core-modules/auth/services/reset-message-folder.service.ts deleted file mode 100644 index 93960cee2f..0000000000 --- a/packages/twenty-server/src/engine/core-modules/auth/services/reset-message-folder.service.ts +++ /dev/null @@ -1,69 +0,0 @@ -import { Injectable } from '@nestjs/common'; - -import { In } from 'typeorm'; - -import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; -import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; -import { type MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity'; - -export type ResetMessageFoldersInput = { - workspaceId: string; - connectedAccountId: string; - manager: WorkspaceEntityManager; -}; - -@Injectable() -export class ResetMessageFolderService { - constructor( - private readonly twentyORMGlobalManager: TwentyORMGlobalManager, - ) {} - - async resetMessageFolders(input: ResetMessageFoldersInput): Promise { - const { workspaceId, connectedAccountId, manager } = input; - - const messageChannelRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - 'messageChannel', - ); - - const messageChannels = await messageChannelRepository.find({ - where: { connectedAccountId }, - }); - - const messageChannelIds = messageChannels.map((channel) => channel.id); - - if (messageChannelIds.length === 0) { - return; - } - - const messageFolderRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - 'messageFolder', - ); - - const messageFolders = await messageFolderRepository.find({ - where: { - messageChannelId: In(messageChannelIds), - }, - }); - - if (messageFolders.length === 0) { - return; - } - - await messageFolderRepository.update( - { - messageChannelId: In(messageChannelIds), - }, - { - syncCursor: '', - }, - manager, - ); - - return; - } -} diff --git a/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts b/packages/twenty-server/src/modules/messaging/common/query-hooks/message/message-channel-update-one.pre-query.hook.ts similarity index 90% rename from packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts rename to packages/twenty-server/src/modules/messaging/common/query-hooks/message/message-channel-update-one.pre-query.hook.ts index 99aa1fdefe..6438f5e533 100644 --- a/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts +++ b/packages/twenty-server/src/modules/messaging/common/query-hooks/message/message-channel-update-one.pre-query.hook.ts @@ -1,7 +1,7 @@ import { Logger } from '@nestjs/common'; import { msg } from '@lingui/core/macro'; -import { assertIsDefinedOrThrow } from 'twenty-shared/utils'; +import { assertIsDefinedOrThrow, isDefined } from 'twenty-shared/utils'; import { Not } from 'typeorm'; import { type WorkspacePreQueryHookInstance } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/interfaces/workspace-query-hook.interface'; @@ -28,7 +28,6 @@ import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/ const ONGOING_SYNC_STAGES = [ MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING, - MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING, ]; @WorkspaceQueryHook(`messageChannel.updateOne`) @@ -63,7 +62,7 @@ export class MessageChannelUpdateOnePreQueryHook where: { id: payload.id }, }); - if (!messageChannel) { + if (!isDefined(messageChannel)) { throw new WorkspaceQueryRunnerException( 'Message channel not found', WorkspaceQueryRunnerExceptionCode.DATA_NOT_FOUND, @@ -83,14 +82,15 @@ export class MessageChannelUpdateOnePreQueryHook 'messageFolder', ); - const folderWithPendingAction = await messageFolderRepository.findOne({ - where: { - messageChannelId: messageChannel.id, - pendingSyncAction: Not(MessageFolderPendingSyncAction.NONE), - }, - }); + const messageFoldersWithPendingActionCount = + await messageFolderRepository.count({ + where: { + messageChannelId: messageChannel.id, + pendingSyncAction: Not(MessageFolderPendingSyncAction.NONE), + }, + }); - const hasPendingFolderActions = !!folderWithPendingAction; + const hasPendingFolderActions = messageFoldersWithPendingActionCount > 0; const hasPendingGroupEmailsAction = messageChannel.pendingGroupEmailsAction !== @@ -122,6 +122,7 @@ export class MessageChannelUpdateOnePreQueryHook } const excludeGroupEmailsChanged = + isDefined(payload.data.excludeGroupEmails) && payload.data.excludeGroupEmails !== messageChannel.excludeGroupEmails; if (excludeGroupEmailsChanged) { diff --git a/packages/twenty-server/src/modules/messaging/common/query-hooks/messaging-query-hook.module.ts b/packages/twenty-server/src/modules/messaging/common/query-hooks/messaging-query-hook.module.ts index 8fc1a60179..7ff63adf4b 100644 --- a/packages/twenty-server/src/modules/messaging/common/query-hooks/messaging-query-hook.module.ts +++ b/packages/twenty-server/src/modules/messaging/common/query-hooks/messaging-query-hook.module.ts @@ -1,15 +1,18 @@ import { Module } from '@nestjs/common'; import { ApplyMessagesVisibilityRestrictionsService } from 'src/modules/messaging/common/query-hooks/message/apply-messages-visibility-restrictions.service'; +import { MessageChannelUpdateOnePreQueryHook } from 'src/modules/messaging/common/query-hooks/message/message-channel-update-one.pre-query.hook'; import { MessageFindManyPostQueryHook } from 'src/modules/messaging/common/query-hooks/message/message-find-many.post-query.hook'; import { MessageFindOnePostQueryHook } from 'src/modules/messaging/common/query-hooks/message/message-find-one.post-query.hook'; +import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module'; @Module({ - imports: [], + imports: [MessagingImportManagerModule], providers: [ ApplyMessagesVisibilityRestrictionsService, MessageFindOnePostQueryHook, MessageFindManyPostQueryHook, + MessageChannelUpdateOnePreQueryHook, ], }) export class MessagingQueryHookModule {} diff --git a/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts b/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts index 094e4858f9..9484e8ddc0 100644 --- a/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts +++ b/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts @@ -1,39 +1,48 @@ import { Injectable } from '@nestjs/common'; -import { Any } from 'typeorm'; +import { Any, In } from 'typeorm'; import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; -import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; +import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { AccountsToReconnectKeys } from 'src/modules/connected-account/types/accounts-to-reconnect-key-value.type'; import { + MessageChannelPendingGroupEmailsAction, MessageChannelSyncStage, MessageChannelSyncStatus, type MessageChannelWorkspaceEntity, } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; +import { + MessageFolderPendingSyncAction, + MessageFolderWorkspaceEntity, +} from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity'; @Injectable() export class MessageChannelSyncStatusService { constructor( @InjectCacheStorage(CacheStorageNamespace.ModuleMessaging) private readonly cacheStorage: CacheStorageService, - private readonly twentyORMManager: TwentyORMManager, + private readonly twentyORMGlobalManager: TwentyORMGlobalManager, private readonly accountsToReconnectService: AccountsToReconnectService, private readonly metricsService: MetricsService, ) {} - public async scheduleMessageListFetch(messageChannelIds: string[]) { + public async scheduleMessageListFetch( + messageChannelIds: string[], + workspaceId: string, + ) { if (!messageChannelIds.length) { return; } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); @@ -42,13 +51,17 @@ export class MessageChannelSyncStatusService { }); } - public async scheduleMessagesImport(messageChannelIds: string[]) { + public async scheduleMessagesImport( + messageChannelIds: string[], + workspaceId: string, + ) { if (!messageChannelIds.length) { return; } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); @@ -72,26 +85,46 @@ export class MessageChannelSyncStatusService { } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); + const messageFolderRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + 'messageFolder', + ); + await messageChannelRepository.update(messageChannelIds, { syncCursor: '', syncStageStartedAt: null, throttleFailureCount: 0, + pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, }); - await this.scheduleMessageListFetch(messageChannelIds); + await messageFolderRepository.update( + { messageChannelId: In(messageChannelIds) }, + { + syncCursor: '', + pendingSyncAction: MessageFolderPendingSyncAction.NONE, + }, + ); + + await this.scheduleMessageListFetch(messageChannelIds, workspaceId); } - public async resetSyncStageStartedAt(messageChannelIds: string[]) { + public async resetSyncStageStartedAt( + messageChannelIds: string[], + workspaceId: string, + ) { if (!messageChannelIds.length) { return; } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); @@ -100,13 +133,17 @@ export class MessageChannelSyncStatusService { }); } - public async markAsMessagesListFetchOngoing(messageChannelIds: string[]) { + public async markAsMessagesListFetchOngoing( + messageChannelIds: string[], + workspaceId: string, + ) { if (!messageChannelIds.length) { return; } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); @@ -119,13 +156,15 @@ export class MessageChannelSyncStatusService { public async markAsCompletedAndScheduleMessageListFetch( messageChannelIds: string[], + workspaceId: string, ) { if (!messageChannelIds.length) { return; } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); @@ -143,13 +182,17 @@ export class MessageChannelSyncStatusService { }); } - public async markAsMessagesImportOngoing(messageChannelIds: string[]) { + public async markAsMessagesImportOngoing( + messageChannelIds: string[], + workspaceId: string, + ) { if (!messageChannelIds.length) { return; } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); @@ -171,7 +214,8 @@ export class MessageChannelSyncStatusService { } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); @@ -194,7 +238,8 @@ export class MessageChannelSyncStatusService { syncStatus === MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS ) { const connectedAccountRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'connectedAccount', ); @@ -230,7 +275,8 @@ export class MessageChannelSyncStatusService { } const messageChannelRepository = - await this.twentyORMManager.getRepository( + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, 'messageChannel', ); diff --git a/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-query-hook.module.ts b/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-query-hook.module.ts deleted file mode 100644 index d0876baaa0..0000000000 --- a/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-query-hook.module.ts +++ /dev/null @@ -1,11 +0,0 @@ -import { Module } from '@nestjs/common'; - -import { MessageChannelUpdateOnePreQueryHook } from 'src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook'; -import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module'; - -@Module({ - imports: [MessagingImportManagerModule], - providers: [MessageChannelUpdateOnePreQueryHook], - exports: [MessageChannelUpdateOnePreQueryHook], -}) -export class MessageChannelQueryHookModule {} diff --git a/packages/twenty-server/src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command.ts b/packages/twenty-server/src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command.ts new file mode 100644 index 0000000000..3d37a87fa6 --- /dev/null +++ b/packages/twenty-server/src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command.ts @@ -0,0 +1,97 @@ +import { Logger } from '@nestjs/common'; + +import { Command, CommandRunner, Option } from 'nest-commander'; +import { isDefined } from 'twenty-shared/utils'; + +import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; +import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; +import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service'; + +type MessagingResetChannelCommandOptions = { + workspaceId: string; + messageChannelId?: string; +}; + +@Command({ + name: 'messaging:reset-channel', + description: + 'Reset message channel(s) for full resync. If no channel ID provided, resets all channels in the workspace.', +}) +export class MessagingResetChannelCommand extends CommandRunner { + private readonly logger = new Logger(MessagingResetChannelCommand.name); + + constructor( + private readonly twentyORMGlobalManager: TwentyORMGlobalManager, + private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService, + private readonly messagingMessageCleanerService: MessagingMessageCleanerService, + ) { + super(); + } + + async run( + _passedParam: string[], + options: MessagingResetChannelCommandOptions, + ): Promise { + const { workspaceId, messageChannelId } = options; + + const messageChannelRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + 'messageChannel', + ); + + this.logger.log( + `No message channel ID provided, resetting all message channels in workspace ${workspaceId}`, + ); + + const messageChannels = await messageChannelRepository.find({ + where: { + ...(isDefined(messageChannelId) ? { id: messageChannelId } : {}), + }, + }); + + if (messageChannels.length === 0) { + this.logger.log(`No message channels found in workspace ${workspaceId}`); + + return; + } + + this.logger.log( + `Found ${messageChannels.length} message channels to reset`, + ); + + for (const messageChannel of messageChannels) { + await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch( + [messageChannel.id], + workspaceId, + ); + await this.messagingMessageCleanerService.cleanOrphanMessagesAndThreads( + workspaceId, + ); + } + + this.logger.log( + `Successfully reset all ${messageChannels.length} message channels in workspace ${workspaceId}`, + ); + } + + @Option({ + flags: '-w, --workspace-id ', + description: 'Workspace ID', + required: true, + }) + parseWorkspaceId(value: string): string { + return value; + } + + @Option({ + flags: '-c, --message-channel-id [message_channel_id]', + description: + 'Message Channel ID (optional - if not provided, all channels will be reset)', + required: false, + }) + parseMessageChannelId(value: string): string { + return value; + } +} diff --git a/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts b/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts index 11ec4212b1..523c51f8db 100644 --- a/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts +++ b/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts @@ -3,18 +3,25 @@ import { TypeOrmModule } from '@nestjs/typeorm'; import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module'; +import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module'; import { MessagingMessageCleanerRemoveOrphansCommand } from 'src/modules/messaging/message-cleaner/commands/messaging-message-clearner-remove-orphans.command'; +import { MessagingResetChannelCommand } from 'src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command'; import { MessagingConnectedAccountDeletionCleanupJob } from 'src/modules/messaging/message-cleaner/jobs/messaging-connected-account-deletion-cleanup.job'; import { MessagingMessageCleanerConnectedAccountListener } from 'src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener'; import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service'; @Module({ - imports: [TypeOrmModule.forFeature([WorkspaceEntity]), DataSourceModule], + imports: [ + TypeOrmModule.forFeature([WorkspaceEntity]), + DataSourceModule, + MessagingCommonModule, + ], providers: [ - MessagingMessageCleanerService, MessagingConnectedAccountDeletionCleanupJob, MessagingMessageCleanerConnectedAccountListener, MessagingMessageCleanerRemoveOrphansCommand, + MessagingResetChannelCommand, + MessagingMessageCleanerService, ], exports: [MessagingMessageCleanerService], }) diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts index 83c8977a40..3741aaecb9 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts @@ -57,22 +57,25 @@ export class MessagingOngoingStaleJob { `Sync for message channel ${messageChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to MESSAGES_IMPORT_PENDING`, ); - await this.messageChannelSyncStatusService.resetSyncStageStartedAt([ - messageChannel.id, - ]); + await this.messageChannelSyncStatusService.resetSyncStageStartedAt( + [messageChannel.id], + workspaceId, + ); switch (messageChannel.syncStage) { case MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING: case MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED: await this.messageChannelSyncStatusService.scheduleMessageListFetch( [messageChannel.id], + workspaceId, ); break; case MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING: case MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED: - await this.messageChannelSyncStatusService.scheduleMessagesImport([ - messageChannel.id, - ]); + await this.messageChannelSyncStatusService.scheduleMessagesImport( + [messageChannel.id], + workspaceId, + ); break; default: break; diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts index 623ebaceb2..51ba0f753e 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts @@ -35,7 +35,6 @@ import { MessagingOngoingStaleJob } from 'src/modules/messaging/message-import-m import { MessagingRelaunchFailedMessageChannelJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-relaunch-failed-message-channel.job'; import { MessagingMessageImportManagerMessageChannelListener } from 'src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener'; import { MessagingAccountAuthenticationService } from 'src/modules/messaging/message-import-manager/services/messaging-account-authentication.service'; -import { MessagingClearCursorsModule } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.module'; import { MessagingCursorService } from 'src/modules/messaging/message-import-manager/services/messaging-cursor.service'; import { MessagingDeleteFolderMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-folder-messages.service'; import { MessagingDeleteGroupEmailMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-group-email-messages.service'; @@ -70,7 +69,6 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess FeatureFlagModule, MessageParticipantManagerModule, MessagingFolderSyncManagerModule, - MessagingClearCursorsModule, MessagingMonitoringModule, MessagingMessageCleanerModule, WorkspaceEventEmitterModule, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.module.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.module.ts deleted file mode 100644 index 72927d99f2..0000000000 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.module.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { Module } from '@nestjs/common'; - -import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service'; - -@Module({ - providers: [MessagingClearCursorsService], - exports: [MessagingClearCursorsService], -}) -export class MessagingClearCursorsModule {} diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.spec.ts deleted file mode 100644 index c5dda9cfa3..0000000000 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.spec.ts +++ /dev/null @@ -1,62 +0,0 @@ -import { Test, type TestingModule } from '@nestjs/testing'; - -import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service'; - -describe('MessagingClearCursorsService', () => { - let service: MessagingClearCursorsService; - - const mockMessageChannelRepository = { - update: jest.fn(), - }; - - const mockMessageFolderRepository = { - update: jest.fn(), - }; - - beforeEach(async () => { - const module: TestingModule = await Test.createTestingModule({ - providers: [ - MessagingClearCursorsService, - { - provide: TwentyORMManager, - useValue: { - getRepository: jest.fn((entityName) => { - if (entityName === 'messageChannel') { - return mockMessageChannelRepository; - } - if (entityName === 'messageFolder') { - return mockMessageFolderRepository; - } - }), - }, - }, - ], - }).compile(); - - service = module.get(MessagingClearCursorsService); - }); - - afterEach(() => { - jest.clearAllMocks(); - }); - - describe('clearAllMessageChannelCursors', () => { - const messageChannelId = 'test-channel-id'; - - it('should clear message channel and folder cursors', async () => { - await service.clearAllMessageChannelCursors(messageChannelId); - - expect(mockMessageChannelRepository.update).toHaveBeenCalledWith( - { id: messageChannelId }, - { syncCursor: '' }, - undefined, - ); - expect(mockMessageFolderRepository.update).toHaveBeenCalledWith( - { messageChannelId }, - { syncCursor: '' }, - undefined, - ); - }); - }); -}); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.ts deleted file mode 100644 index feabc194b7..0000000000 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.ts +++ /dev/null @@ -1,48 +0,0 @@ -import { Injectable, Logger } from '@nestjs/common'; - -import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; -import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; -import { MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity'; - -@Injectable() -export class MessagingClearCursorsService { - private readonly logger = new Logger(MessagingClearCursorsService.name); - - constructor(private readonly twentyORMManager: TwentyORMManager) {} - - async clearAllMessageChannelCursors( - messageChannelId: string, - transactionManager?: WorkspaceEntityManager, - ): Promise { - this.logger.log( - `MessageChannelId: ${messageChannelId} - Clearing all sync cursors`, - ); - - const messageChannelRepository = - await this.twentyORMManager.getRepository( - 'messageChannel', - ); - - const messageFolderRepository = - await this.twentyORMManager.getRepository( - 'messageFolder', - ); - - await messageChannelRepository.update( - { id: messageChannelId }, - { syncCursor: '' }, - transactionManager, - ); - - await messageFolderRepository.update( - { messageChannelId }, - { syncCursor: '' }, - transactionManager, - ); - - this.logger.log( - `MessageChannelId: ${messageChannelId} - Cleared all sync cursors`, - ); - } -} diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts index 1952c16e15..281551728e 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts @@ -17,7 +17,6 @@ import { MessageImportDriverExceptionCode, } from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception'; import { MessageNetworkExceptionCode } from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-network.exception'; -import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service'; export enum MessageImportSyncStep { MESSAGE_LIST_FETCH = 'MESSAGE_LIST_FETCH', @@ -30,7 +29,6 @@ export class MessageImportExceptionHandlerService { constructor( private readonly twentyORMManager: TwentyORMManager, private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService, - private readonly messagingClearCursorsService: MessagingClearCursorsService, private readonly exceptionHandlerService: ExceptionHandlerService, ) {} @@ -113,10 +111,6 @@ export class MessageImportExceptionHandlerService { workspaceId, MessageChannelSyncStatus.FAILED_UNKNOWN, ); - - await this.messagingClearCursorsService.clearAllMessageChannelCursors( - messageChannel.id, - ); } private async handleTemporaryException( @@ -169,16 +163,18 @@ export class MessageImportExceptionHandlerService { switch (syncStep) { case MessageImportSyncStep.MESSAGE_LIST_FETCH: - await this.messageChannelSyncStatusService.scheduleMessageListFetch([ - messageChannel.id, - ]); + await this.messageChannelSyncStatusService.scheduleMessageListFetch( + [messageChannel.id], + workspaceId, + ); break; case MessageImportSyncStep.MESSAGES_IMPORT_PENDING: case MessageImportSyncStep.MESSAGES_IMPORT_ONGOING: - await this.messageChannelSyncStatusService.scheduleMessagesImport([ - messageChannel.id, - ]); + await this.messageChannelSyncStatusService.scheduleMessagesImport( + [messageChannel.id], + workspaceId, + ); break; default: diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts index ae17f908d6..248e463b00 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts @@ -67,6 +67,7 @@ export class MessagingMessageListFetchService { await this.messageChannelSyncStatusService.markAsMessagesListFetchOngoing( [messageChannel.id], + workspaceId, ); this.logger.log( @@ -270,6 +271,7 @@ export class MessagingMessageListFetchService { if (totalMessagesToImportCount === 0) { await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch( [messageChannelWithFreshTokens.id], + workspaceId, ); return; @@ -279,9 +281,10 @@ export class MessagingMessageListFetchService { `messageChannelId: ${freshMessageChannel.id} Scheduling direct messages import`, ); - await this.messageChannelSyncStatusService.scheduleMessagesImport([ - messageChannelWithFreshTokens.id, - ]); + await this.messageChannelSyncStatusService.scheduleMessagesImport( + [messageChannelWithFreshTokens.id], + workspaceId, + ); await this.messagingMessagesImportService.processMessageBatchImport( { diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts index 7189a7901c..cc6bdf79ca 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts @@ -71,9 +71,10 @@ export class MessagingMessagesImportService { messageChannelId: messageChannel.id, }); - await this.messageChannelSyncStatusService.markAsMessagesImportOngoing([ - messageChannel.id, - ]); + await this.messageChannelSyncStatusService.markAsMessagesImportOngoing( + [messageChannel.id], + workspaceId, + ); const { accessToken, refreshToken } = await this.messagingAccountAuthenticationService.validateAndRefreshConnectedAccountAuthentication( @@ -102,6 +103,7 @@ export class MessagingMessagesImportService { if (!messageIdsToFetch?.length) { await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch( [messageChannel.id], + workspaceId, ); return await this.trackMessageImportCompleted( @@ -158,11 +160,13 @@ export class MessagingMessagesImportService { ) { await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch( [messageChannel.id], + workspaceId, ); } else { - await this.messageChannelSyncStatusService.scheduleMessagesImport([ - messageChannel.id, - ]); + await this.messageChannelSyncStatusService.scheduleMessagesImport( + [messageChannel.id], + workspaceId, + ); } const messageChannelRepository = diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts index 674f63d77a..529e46a2a9 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts @@ -8,7 +8,7 @@ import { MessageChannelPendingGroupEmailsAction, MessageChannelWorkspaceEntity, } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; -import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service'; +import { MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity'; import { MessagingDeleteGroupEmailMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-group-email-messages.service'; @Injectable() @@ -20,7 +20,6 @@ export class MessagingProcessGroupEmailActionsService { constructor( private readonly twentyORMManager: TwentyORMManager, private readonly messagingDeleteGroupEmailMessagesService: MessagingDeleteGroupEmailMessagesService, - private readonly messagingClearCursorsService: MessagingClearCursorsService, ) {} async markMessageChannelAsPendingGroupEmailsAction( @@ -120,10 +119,10 @@ export class MessagingProcessGroupEmailActionsService { messageChannelId, ); - await this.messagingClearCursorsService.clearAllMessageChannelCursors( + await this.resetCursors({ messageChannelId, transactionManager, - ); + }); this.logger.log( `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Completed GROUP_EMAILS_DELETION action`, @@ -135,13 +134,45 @@ export class MessagingProcessGroupEmailActionsService { messageChannelId: string, transactionManager: WorkspaceEntityManager, ): Promise { - await this.messagingClearCursorsService.clearAllMessageChannelCursors( + await this.resetCursors({ messageChannelId, transactionManager, - ); + }); this.logger.log( `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Completed GROUP_EMAILS_IMPORT action`, ); } + + private async resetCursors({ + messageChannelId, + transactionManager, + }: { + messageChannelId: string; + transactionManager: WorkspaceEntityManager; + }) { + const messageChannelRepository = + await this.twentyORMManager.getRepository( + 'messageChannel', + ); + + await messageChannelRepository.update( + messageChannelId, + { + syncCursor: '', + }, + transactionManager, + ); + + const messageFolderRepository = + await this.twentyORMManager.getRepository( + 'messageFolder', + ); + + await messageFolderRepository.update( + { messageChannelId }, + { syncCursor: '' }, + transactionManager, + ); + } } diff --git a/packages/twenty-server/src/modules/messaging/messaging.module.ts b/packages/twenty-server/src/modules/messaging/messaging.module.ts index 50c605f369..ed25046ae2 100644 --- a/packages/twenty-server/src/modules/messaging/messaging.module.ts +++ b/packages/twenty-server/src/modules/messaging/messaging.module.ts @@ -1,7 +1,6 @@ import { Module } from '@nestjs/common'; import { MessagingBlocklistManagerModule } from 'src/modules/messaging/blocklist-manager/messaging-blocklist-manager.module'; -import { MessageChannelQueryHookModule } from 'src/modules/messaging/message-channel-manager/query-hooks/message-channel-query-hook.module'; import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cleaner/messaging-message-cleaner.module'; import { MessageFolderQueryHookModule } from 'src/modules/messaging/message-folder-manager/query-hooks/message-folder-query-hook.module'; import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module'; @@ -15,7 +14,6 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess MessageParticipantManagerModule, MessagingBlocklistManagerModule, MessagingMonitoringModule, - MessageChannelQueryHookModule, MessageFolderQueryHookModule, ], providers: [],