feat: migrate ConnectedAccount infrastructure entities to metadata schema (#18784)

## Summary

- Migrates 4 entities (`connectedAccount`, `messageChannel`,
`calendarChannel`, `messageFolder`) from per-workspace schemas to the
shared `core` metadata schema
- Introduces a `IS_CONNECTED_ACCOUNT_MIGRATED` feature flag to control
the migration: when enabled, reads come from core metadata and all
writes are dual-written to both workspace and core
- Extracts 12 enums from workspace entity files to `twenty-shared` for
reuse across frontend and backend
- Creates new TypeORM entities, metadata services, GraphQL
resolvers/DTOs, and exception interceptors per entity
- Each entity owns its own data access module
(`ConnectedAccountDataAccessModule`, `MessageChannelDataAccessModule`,
`CalendarChannelDataAccessModule`, `MessageFolderDataAccessModule`) — no
umbrella infrastructure module
- Adds a 1.20 upgrade command that backfills data from workspace schemas
to core (preserving UUIDs) and enables the feature flag
- Replaces direct repository access with data access service calls
across ~50 files in messaging, calendar, and connected-account modules
- Adds `lastSignedInAt` and `oidcTokenClaims` fields to the new
`ConnectedAccountEntity`
- Drops unused `lastSyncHistoryId` field from the migrated connected
account entity

## Test plan

- [x] Lint passes (`npx nx lint:diff-with-main twenty-server`)
- [x] Typecheck passes (`npx nx typecheck twenty-server`)
- [x] All unit tests pass (477 suites, 4267 tests, 0 failures)
- [ ] Manual test: verify messaging sync works with feature flag
disabled (existing behavior)
- [ ] Manual test: run upgrade command on a workspace, verify data
backfilled to core tables
- [ ] Manual test: verify messaging/calendar sync works with feature
flag enabled (dual-write path)
- [ ] Manual test: verify GraphQL metadata resolvers return correct data
when flag enabled

packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts

@@ -1,19 +1,30 @@
 import { Scope } from '@nestjs/common';
 
+import { type ObjectRecordCreateEvent } from 'twenty-shared/database-events';
 import { MessageParticipantRole } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';
-import { And, Any, ILike, In, IsNull, Not, Or } from 'typeorm';
-import { type ObjectRecordCreateEvent } from 'twenty-shared/database-events';
+import {
+  And,
+  Any,
+  type FindManyOptions,
+  ILike,
+  In,
+  IsNull,
+  Not,
+  Or,
+} from 'typeorm';
+
+import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
 import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
 import { type MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
-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';
 
 export type BlocklistItemDeleteMessagesJobData = WorkspaceEventBatch<
@@ -28,6 +39,7 @@ export class BlocklistItemDeleteMessagesJob {
   constructor(
     private readonly threadCleanerService: MessagingMessageCleanerService,
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
   ) {}
 
   @Process(BlocklistItemDeleteMessagesJob.name)
@@ -72,12 +84,6 @@ export class BlocklistItemDeleteMessagesJob {
         new Map<string, string[]>(),
       );
 
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
       const messageChannelMessageAssociationRepository =
         await this.globalWorkspaceOrmManager.getRepository<MessageChannelMessageAssociationWorkspaceEntity>(
           workspaceId,
@@ -97,30 +103,35 @@ export class BlocklistItemDeleteMessagesJob {
           MessageParticipantRole.TO,
         ] as const;
 
-        const messageChannels = await messageChannelRepository.find({
-          select: {
-            id: true,
-            handle: true,
-            connectedAccount: {
-              handleAliases: true,
+        const messageChannels =
+          await this.messageChannelDataAccessService.findMany(workspaceId, {
+            select: {
+              id: true,
+              handle: true,
+              connectedAccount: {
+                handleAliases: true,
+              },
             },
-          },
-          where: {
-            connectedAccount: {
-              accountOwnerId: workspaceMemberId,
-              deletedAt: IsNull(),
+            where: {
+              connectedAccount: {
+                accountOwnerId: workspaceMemberId,
+                deletedAt: IsNull(),
+              },
             },
-          },
-          relations: ['connectedAccount'],
-        });
+            relations: ['connectedAccount'],
+          } as FindManyOptions<MessageChannelWorkspaceEntity>);
 
         for (const messageChannel of messageChannels) {
           const messageChannelHandles = [messageChannel.handle];
 
-          if (isDefined(messageChannel.connectedAccount.handleAliases)) {
-            messageChannelHandles.push(
-              ...messageChannel.connectedAccount.handleAliases.split(','),
-            );
+          const handleAliases = messageChannel.connectedAccount?.handleAliases;
+
+          if (isDefined(handleAliases)) {
+            const aliasList: string[] = Array.isArray(handleAliases)
+              ? handleAliases
+              : (handleAliases as string).split(',');
+
+            messageChannelHandles.push(...aliasList);
           }
 
           const handleConditions = handles.map((handle) => {

packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts

@@ -6,15 +6,13 @@ import { type ObjectRecordDeleteEvent } from 'twenty-shared/database-events';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
 import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
 import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
-import {
-  MessageChannelSyncStage,
-  type MessageChannelWorkspaceEntity,
-} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { MessageChannelSyncStage } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 
 export type BlocklistReimportMessagesJobData = WorkspaceEventBatch<
   ObjectRecordDeleteEvent<BlocklistWorkspaceEntity>
@@ -27,6 +25,7 @@ export type BlocklistReimportMessagesJobData = WorkspaceEventBatch<
 export class BlocklistReimportMessagesJob {
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService,
   ) {}
 
@@ -37,25 +36,19 @@ export class BlocklistReimportMessagesJob {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
       for (const eventPayload of data.events) {
         const workspaceMemberId =
           eventPayload.properties.before.workspaceMemberId;
 
-        const messageChannels = await messageChannelRepository.find({
-          select: ['id'],
-          where: {
+        const messageChannels = await this.messageChannelDataAccessService.find(
+          workspaceId,
+          {
             connectedAccount: {
               accountOwnerId: workspaceMemberId,
             },
             syncStage: Not(MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING),
           },
-        });
+        );
 
         await this.messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
           messageChannels.map((messageChannel) => messageChannel.id),

packages/twenty-server/src/modules/messaging/blocklist-manager/messaging-blocklist-manager.module.ts

@@ -1,5 +1,6 @@
 import { Module } from '@nestjs/common';
 
+import { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module';
 import { BlocklistItemDeleteMessagesJob } from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job';
 import { BlocklistReimportMessagesJob } from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job';
 import { MessagingBlocklistListener } from 'src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener';
@@ -7,7 +8,11 @@ import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-co
 import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cleaner/messaging-message-cleaner.module';
 
 @Module({
-  imports: [MessagingCommonModule, MessagingMessageCleanerModule],
+  imports: [
+    MessagingCommonModule,
+    MessagingMessageCleanerModule,
+    MessageChannelDataAccessModule,
+  ],
   providers: [
     MessagingBlocklistListener,
     BlocklistItemDeleteMessagesJob,

packages/twenty-server/src/modules/messaging/common/messaging-common.module.ts

@@ -3,6 +3,9 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 
 import { FeatureFlagEntity } from 'src/engine/core-modules/feature-flag/feature-flag.entity';
 import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
+import { ConnectedAccountDataAccessModule } from 'src/engine/metadata-modules/connected-account/data-access/connected-account-data-access.module';
+import { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module';
+import { MessageFolderDataAccessModule } from 'src/engine/metadata-modules/message-folder/data-access/message-folder-data-access.module';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { ConnectedAccountModule } from 'src/modules/connected-account/connected-account.module';
 import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
@@ -12,6 +15,9 @@ import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/se
     WorkspaceDataSourceModule,
     TypeOrmModule.forFeature([FeatureFlagEntity]),
     ConnectedAccountModule,
+    ConnectedAccountDataAccessModule,
+    MessageChannelDataAccessModule,
+    MessageFolderDataAccessModule,
     MetricsModule,
   ],
   providers: [MessageChannelSyncStatusService],

packages/twenty-server/src/modules/messaging/common/query-hooks/message/apply-messages-visibility-restrictions.service.spec.ts

@@ -2,6 +2,7 @@ import { Test, type TestingModule } from '@nestjs/testing';
 
 import { FIELD_RESTRICTED_ADDITIONAL_PERMISSIONS_REQUIRED } from 'twenty-shared/constants';
 
+import { ConnectedAccountDataAccessService } from 'src/engine/metadata-modules/connected-account/data-access/services/connected-account-data-access.service';
 import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
 import { MessageChannelVisibility } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import { type MessageWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message.workspace-entity';
@@ -34,22 +35,19 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
     find: jest.fn(),
   };
 
-  const mockConnectedAccountRepository = {
-    find: jest.fn(),
-  };
-
   const mockWorkspaceMemberRepository = {
     findOneByOrFail: jest.fn(),
   };
 
+  const mockConnectedAccountDataAccessService = {
+    find: jest.fn(),
+  };
+
   const mockGlobalWorkspaceOrmManager = {
     getRepository: jest.fn().mockImplementation((workspaceId, name) => {
       if (name === 'messageChannelMessageAssociation') {
         return mockMessageChannelMessageAssociationRepository;
       }
-      if (name === 'connectedAccount') {
-        return mockConnectedAccountRepository;
-      }
       if (name === 'workspaceMember') {
         return mockWorkspaceMemberRepository;
       }
@@ -67,6 +65,10 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
           provide: GlobalWorkspaceOrmManager,
           useValue: mockGlobalWorkspaceOrmManager,
         },
+        {
+          provide: ConnectedAccountDataAccessService,
+          useValue: mockConnectedAccountDataAccessService,
+        },
       ],
     }).compile();
 
@@ -105,7 +107,7 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
           item.subject === 'Test Subject' && item.text === 'Test Message',
       ),
     ).toBe(true);
-    expect(mockConnectedAccountRepository.find).not.toHaveBeenCalled();
+    expect(mockConnectedAccountDataAccessService.find).not.toHaveBeenCalled();
   });
 
   it('should return message without obfuscated subject and with obfuscated text if the visibility is SUBJECT', async () => {
@@ -123,7 +125,7 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
       },
     ]);
 
-    mockConnectedAccountRepository.find.mockResolvedValue([]);
+    mockConnectedAccountDataAccessService.find.mockResolvedValue([]);
 
     mockWorkspaceMemberRepository.findOneByOrFail.mockResolvedValue({
       id: 'workspace-member-id',
@@ -158,7 +160,7 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
       },
     ]);
 
-    mockConnectedAccountRepository.find.mockResolvedValue([]);
+    mockConnectedAccountDataAccessService.find.mockResolvedValue([]);
 
     mockWorkspaceMemberRepository.findOneByOrFail.mockResolvedValue({
       id: 'workspace-member-id',
@@ -198,7 +200,7 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
       id: 'workspace-member-account-owner-id',
     });
 
-    mockConnectedAccountRepository.find.mockResolvedValue([{ id: '1' }]);
+    mockConnectedAccountDataAccessService.find.mockResolvedValue([{ id: '1' }]);
 
     const result = await service.applyMessagesVisibilityRestrictions(
       messages,
@@ -233,7 +235,7 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
       id: 'workspace-member-not-account-owner-id',
     });
 
-    mockConnectedAccountRepository.find.mockResolvedValue([]);
+    mockConnectedAccountDataAccessService.find.mockResolvedValue([]);
 
     const result = await service.applyMessagesVisibilityRestrictions(
       messages,
@@ -279,7 +281,7 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
       id: 'workspace-member-id',
     });
 
-    mockConnectedAccountRepository.find
+    mockConnectedAccountDataAccessService.find
       .mockResolvedValueOnce([]) // request for message 3
       .mockResolvedValueOnce([]); // request for message 2
 
@@ -338,7 +340,7 @@ describe('ApplyMessagesVisibilityRestrictionsService', () => {
       id: 'workspace-member-id',
     });
 
-    mockConnectedAccountRepository.find
+    mockConnectedAccountDataAccessService.find
       .mockResolvedValueOnce([]) // request for message 3
       .mockResolvedValueOnce([]); // request for message 2
 

packages/twenty-server/src/modules/messaging/common/query-hooks/message/apply-messages-visibility-restrictions.service.ts

@@ -6,9 +6,9 @@ import { isDefined } from 'twenty-shared/utils';
 import { In } from 'typeorm';
 
 import { NotFoundError } from 'src/engine/core-modules/graphql/utils/graphql-errors.util';
+import { ConnectedAccountDataAccessService } from 'src/engine/metadata-modules/connected-account/data-access/services/connected-account-data-access.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 { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
 import { type MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
 import { MessageChannelVisibility } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import { type MessageWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message.workspace-entity';
@@ -18,6 +18,7 @@ import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/sta
 export class ApplyMessagesVisibilityRestrictionsService {
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly connectedAccountDataAccessService: ConnectedAccountDataAccessService,
   ) {}
 
   public async applyMessagesVisibilityRestrictions(
@@ -43,12 +44,6 @@ export class ApplyMessagesVisibilityRestrictionsService {
             relations: ['messageChannel'],
           });
 
-        const connectedAccountRepository =
-          await this.globalWorkspaceOrmManager.getRepository<ConnectedAccountWorkspaceEntity>(
-            workspaceId,
-            'connectedAccount',
-          );
-
         const workspaceMemberRepository =
           await this.globalWorkspaceOrmManager.getRepository<WorkspaceMemberWorkspaceEntity>(
             workspaceId,
@@ -91,15 +86,13 @@ export class ApplyMessagesVisibilityRestrictionsService {
                 userId,
               });
 
-            const connectedAccounts = await connectedAccountRepository.find({
-              select: ['id'],
-              where: {
+            const connectedAccounts =
+              await this.connectedAccountDataAccessService.find(workspaceId, {
                 messageChannels: {
                   id: In(messageChannels.map((channel) => channel.id)),
                 },
                 accountOwnerId: workspaceMember.id,
-              },
-            });
+              });
 
             if (connectedAccounts.length > 0) {
               continue;

packages/twenty-server/src/modules/messaging/common/query-hooks/message/message-channel-update-one.pre-query.hook.ts

@@ -14,6 +14,8 @@ import {
 } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-runner.exception';
 import { type WorkspaceAuthContext } from 'src/engine/core-modules/auth/types/workspace-auth-context.type';
 import { WorkspaceNotFoundDefaultError } from 'src/engine/core-modules/workspace/workspace.exception';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
+import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.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 {
@@ -21,10 +23,7 @@ import {
   MessageChannelSyncStage,
   type MessageChannelWorkspaceEntity,
 } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
-import {
-  MessageFolderPendingSyncAction,
-  type MessageFolderWorkspaceEntity,
-} from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
+import { MessageFolderPendingSyncAction } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
 import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service';
 
 const ONGOING_SYNC_STAGES = [
@@ -41,6 +40,8 @@ export class MessageChannelUpdateOnePreQueryHook
 
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
+    private readonly messageFolderDataAccessService: MessageFolderDataAccessService,
     private readonly messagingProcessGroupEmailActionsService: MessagingProcessGroupEmailActionsService,
   ) {}
 
@@ -57,15 +58,10 @@ export class MessageChannelUpdateOnePreQueryHook
 
     return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
       async () => {
-        const messageChannelRepository =
-          await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-            workspace.id,
-            'messageChannel',
-          );
-
-        const messageChannel = await messageChannelRepository.findOne({
-          where: { id: payload.id },
-        });
+        const messageChannel =
+          await this.messageChannelDataAccessService.findOne(workspace.id, {
+            where: { id: payload.id },
+          });
 
         if (!isDefined(messageChannel)) {
           throw new WorkspaceQueryRunnerException(
@@ -77,29 +73,27 @@ export class MessageChannelUpdateOnePreQueryHook
           );
         }
 
+        const messageChannelWorkspace =
+          messageChannel as unknown as MessageChannelWorkspaceEntity;
+
         const isSyncOngoing = ONGOING_SYNC_STAGES.includes(
-          messageChannel.syncStage,
+          messageChannelWorkspace.syncStage,
         );
 
-        const messageFolderRepository =
-          await this.globalWorkspaceOrmManager.getRepository<MessageFolderWorkspaceEntity>(
-            workspace.id,
-            'messageFolder',
-          );
+        const messageFoldersWithPendingAction =
+          await this.messageFolderDataAccessService.find(workspace.id, {
+            messageChannelId: messageChannel.id,
+            pendingSyncAction: Not(MessageFolderPendingSyncAction.NONE),
+          });
 
         const messageFoldersWithPendingActionCount =
-          await messageFolderRepository.count({
-            where: {
-              messageChannelId: messageChannel.id,
-              pendingSyncAction: Not(MessageFolderPendingSyncAction.NONE),
-            },
-          });
+          messageFoldersWithPendingAction.length;
 
         const hasPendingFolderActions =
           messageFoldersWithPendingActionCount > 0;
 
         const hasPendingGroupEmailsAction =
-          messageChannel.pendingGroupEmailsAction !==
+          messageChannelWorkspace.pendingGroupEmailsAction !==
           MessageChannelPendingGroupEmailsAction.NONE;
 
         if (
@@ -116,12 +110,12 @@ export class MessageChannelUpdateOnePreQueryHook
         }
 
         const hasCompletedConfiguration =
-          messageChannel.syncStage !==
+          messageChannelWorkspace.syncStage !==
           MessageChannelSyncStage.PENDING_CONFIGURATION;
 
         if (!hasCompletedConfiguration) {
           this.logger.log(
-            `MessageChannelId: ${messageChannel.id} - Skipping pending action for message channel in PENDING_CONFIGURATION state`,
+            `MessageChannelId: ${messageChannelWorkspace.id} - Skipping pending action for message channel in PENDING_CONFIGURATION state`,
           );
 
           return payload;
@@ -129,11 +123,12 @@ export class MessageChannelUpdateOnePreQueryHook
 
         const excludeGroupEmailsChanged =
           isDefined(payload.data.excludeGroupEmails) &&
-          payload.data.excludeGroupEmails !== messageChannel.excludeGroupEmails;
+          payload.data.excludeGroupEmails !==
+            messageChannelWorkspace.excludeGroupEmails;
 
         if (excludeGroupEmailsChanged) {
           await this.messagingProcessGroupEmailActionsService.markMessageChannelAsPendingGroupEmailsAction(
-            messageChannel,
+            messageChannelWorkspace,
             workspace.id,
             payload.data.excludeGroupEmails
               ? MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_DELETION

packages/twenty-server/src/modules/messaging/common/query-hooks/messaging-query-hook.module.ts

@@ -1,5 +1,8 @@
 import { Module } from '@nestjs/common';
 
+import { ConnectedAccountDataAccessModule } from 'src/engine/metadata-modules/connected-account/data-access/connected-account-data-access.module';
+import { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module';
+import { MessageFolderDataAccessModule } from 'src/engine/metadata-modules/message-folder/data-access/message-folder-data-access.module';
 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';
@@ -7,7 +10,12 @@ import { MessageFindOnePostQueryHook } from 'src/modules/messaging/common/query-
 import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
 
 @Module({
-  imports: [MessagingImportManagerModule],
+  imports: [
+    MessagingImportManagerModule,
+    ConnectedAccountDataAccessModule,
+    MessageChannelDataAccessModule,
+    MessageFolderDataAccessModule,
+  ],
   providers: [
     ApplyMessagesVisibilityRestrictionsService,
     MessageFindOnePostQueryHook,

packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts

@@ -7,10 +7,12 @@ import { CacheStorageService } from 'src/engine/core-modules/cache-storage/servi
 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 { ConnectedAccountDataAccessService } from 'src/engine/metadata-modules/connected-account/data-access/services/connected-account-data-access.service';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
+import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.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 { 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,
@@ -18,10 +20,8 @@ import {
   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';
+import { MessageFolderPendingSyncAction } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
+import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
 
 @Injectable()
 export class MessageChannelSyncStatusService {
@@ -29,6 +29,9 @@ export class MessageChannelSyncStatusService {
     @InjectCacheStorage(CacheStorageNamespace.ModuleMessaging)
     private readonly cacheStorage: CacheStorageService,
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
+    private readonly messageFolderDataAccessService: MessageFolderDataAccessService,
+    private readonly connectedAccountDataAccessService: ConnectedAccountDataAccessService,
     private readonly accountsToReconnectService: AccountsToReconnectService,
     private readonly metricsService: MetricsService,
   ) {}
@@ -45,16 +48,14 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
-        ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+          ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
+        },
+      );
     }, authContext);
   }
 
@@ -70,16 +71,14 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
-        ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
+          ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
+        },
+      );
     }, authContext);
   }
 
@@ -100,27 +99,20 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncCursor: '',
+          syncStageStartedAt: null,
+          throttleFailureCount: 0,
+          throttleRetryAfter: null,
+          pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE,
+        },
+      );
 
-      const messageFolderRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageFolderWorkspaceEntity>(
-          workspaceId,
-          'messageFolder',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncCursor: '',
-        syncStageStartedAt: null,
-        throttleFailureCount: 0,
-        throttleRetryAfter: null,
-        pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE,
-      });
-
-      await messageFolderRepository.update(
+      await this.messageFolderDataAccessService.update(
+        workspaceId,
         { messageChannelId: In(messageChannelIds) },
         {
           syncCursor: '',
@@ -143,15 +135,11 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStageStartedAt: null,
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        { syncStageStartedAt: null },
+      );
     }, authContext);
   }
 
@@ -166,17 +154,15 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
-        syncStatus: MessageChannelSyncStatus.ONGOING,
-        syncStageStartedAt: new Date().toISOString(),
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
+          syncStatus: MessageChannelSyncStatus.ONGOING,
+          syncStageStartedAt: new Date().toISOString(),
+        },
+      );
     }, authContext);
   }
 
@@ -191,17 +177,15 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
-        syncStatus: MessageChannelSyncStatus.ONGOING,
-        syncStageStartedAt: new Date().toISOString(),
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
+          syncStatus: MessageChannelSyncStatus.ONGOING,
+          syncStageStartedAt: new Date().toISOString(),
+        },
+      );
     }, authContext);
   }
 
@@ -216,20 +200,18 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStatus: MessageChannelSyncStatus.ACTIVE,
-        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
-        throttleFailureCount: 0,
-        throttleRetryAfter: null,
-        syncStageStartedAt: null,
-        syncedAt: new Date().toISOString(),
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStatus: MessageChannelSyncStatus.ACTIVE,
+          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+          throttleFailureCount: 0,
+          throttleRetryAfter: null,
+          syncStageStartedAt: null,
+          syncedAt: new Date().toISOString(),
+        },
+      );
     }, authContext);
 
     await this.metricsService.batchIncrementCounter({
@@ -249,15 +231,13 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED,
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED,
+        },
+      );
     }, authContext);
   }
 
@@ -272,17 +252,15 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING,
-        syncStatus: MessageChannelSyncStatus.ONGOING,
-        syncStageStartedAt: new Date().toISOString(),
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING,
+          syncStatus: MessageChannelSyncStatus.ONGOING,
+          syncStageStartedAt: new Date().toISOString(),
+        },
+      );
     }, authContext);
   }
 
@@ -300,17 +278,15 @@ export class MessageChannelSyncStatusService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(messageChannelIds, {
-        syncStage: MessageChannelSyncStage.FAILED,
-        syncStatus: syncStatus,
-        throttleRetryAfter: null,
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: In(messageChannelIds) },
+        {
+          syncStage: MessageChannelSyncStage.FAILED,
+          syncStatus: syncStatus,
+          throttleRetryAfter: null,
+        },
+      );
 
       const metricsKey =
         syncStatus === MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS
@@ -325,22 +301,17 @@ export class MessageChannelSyncStatusService {
       if (
         syncStatus === MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS
       ) {
-        const connectedAccountRepository =
-          await this.globalWorkspaceOrmManager.getRepository<ConnectedAccountWorkspaceEntity>(
-            workspaceId,
-            'connectedAccount',
-          );
-
-        const messageChannels = await messageChannelRepository.find({
-          select: ['id', 'connectedAccountId'],
-          where: { id: Any(messageChannelIds) },
-        });
+        const messageChannels = await this.messageChannelDataAccessService.find(
+          workspaceId,
+          { id: In(messageChannelIds) },
+        );
 
         const connectedAccountIds = messageChannels.map(
           (messageChannel) => messageChannel.connectedAccountId,
         );
 
-        await connectedAccountRepository.update(
+        await this.connectedAccountDataAccessService.update(
+          workspaceId,
           { id: Any(connectedAccountIds) },
           {
             authFailedAt: new Date(),
@@ -363,30 +334,43 @@ export class MessageChannelSyncStatusService {
       return;
     }
 
-    const messageChannelRepository =
-      await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
+    const messageChannels = await this.messageChannelDataAccessService.findMany(
+      workspaceId,
+      {
+        where: { id: In(messageChannelIds) },
+      },
+    );
+
+    const workspaceMemberRepository =
+      await this.globalWorkspaceOrmManager.getRepository<WorkspaceMemberWorkspaceEntity>(
         workspaceId,
-        'messageChannel',
+        'workspaceMember',
+        { shouldBypassPermissionChecks: true },
       );
 
-    const messageChannels = await messageChannelRepository.find({
-      where: { id: Any(messageChannelIds) },
-      relations: {
-        connectedAccount: {
-          accountOwner: true,
-        },
-      },
-    });
-
     for (const messageChannel of messageChannels) {
-      const userId = messageChannel.connectedAccount.accountOwner.userId;
-      const connectedAccountId = messageChannel.connectedAccount.id;
+      const connectedAccount =
+        await this.connectedAccountDataAccessService.findOne(workspaceId, {
+          where: { id: messageChannel.connectedAccountId },
+        });
+
+      if (!connectedAccount) {
+        continue;
+      }
+
+      const workspaceMember = await workspaceMemberRepository.findOne({
+        where: { id: connectedAccount.accountOwnerId },
+      });
+
+      if (!workspaceMember) {
+        continue;
+      }
 
       await this.accountsToReconnectService.addAccountToReconnectByKey(
         AccountsToReconnectKeys.ACCOUNTS_TO_RECONNECT_INSUFFICIENT_PERMISSIONS,
-        userId,
+        workspaceMember.userId,
         workspaceId,
-        connectedAccountId,
+        connectedAccount.id,
       );
     }
   }

packages/twenty-server/src/modules/messaging/common/standard-objects/message-channel.workspace-entity.ts

@@ -1,6 +1,15 @@
 import { registerEnumType } from '@nestjs/graphql';
 
-import { FieldMetadataType } from 'twenty-shared/types';
+import {
+  FieldMetadataType,
+  MessageChannelContactAutoCreationPolicy,
+  MessageChannelPendingGroupEmailsAction,
+  MessageChannelSyncStage,
+  MessageChannelSyncStatus,
+  MessageChannelType,
+  MessageChannelVisibility,
+  MessageFolderImportPolicy,
+} from 'twenty-shared/types';
 
 import { BaseWorkspaceEntity } from 'src/engine/twenty-orm/base.workspace-entity';
 import { type FieldTypeAndNameMetadata } from 'src/engine/workspace-manager/utils/get-ts-vector-column-expression.util';
@@ -9,52 +18,15 @@ import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-acco
 import { type MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
 import { type MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
 
-export enum MessageChannelSyncStatus {
-  NOT_SYNCED = 'NOT_SYNCED',
-  ONGOING = 'ONGOING',
-  ACTIVE = 'ACTIVE',
-  FAILED_INSUFFICIENT_PERMISSIONS = 'FAILED_INSUFFICIENT_PERMISSIONS',
-  FAILED_UNKNOWN = 'FAILED_UNKNOWN',
-}
-
-export enum MessageChannelSyncStage {
-  PENDING_CONFIGURATION = 'PENDING_CONFIGURATION',
-  MESSAGE_LIST_FETCH_PENDING = 'MESSAGE_LIST_FETCH_PENDING',
-  MESSAGE_LIST_FETCH_SCHEDULED = 'MESSAGE_LIST_FETCH_SCHEDULED',
-  MESSAGE_LIST_FETCH_ONGOING = 'MESSAGE_LIST_FETCH_ONGOING',
-  MESSAGES_IMPORT_PENDING = 'MESSAGES_IMPORT_PENDING',
-  MESSAGES_IMPORT_SCHEDULED = 'MESSAGES_IMPORT_SCHEDULED',
-  MESSAGES_IMPORT_ONGOING = 'MESSAGES_IMPORT_ONGOING',
-  FAILED = 'FAILED',
-}
-
-export enum MessageChannelVisibility {
-  METADATA = 'METADATA',
-  SUBJECT = 'SUBJECT',
-  SHARE_EVERYTHING = 'SHARE_EVERYTHING',
-}
-
-export enum MessageChannelType {
-  EMAIL = 'EMAIL',
-  SMS = 'SMS',
-}
-
-export enum MessageChannelContactAutoCreationPolicy {
-  SENT_AND_RECEIVED = 'SENT_AND_RECEIVED',
-  SENT = 'SENT',
-  NONE = 'NONE',
-}
-
-export enum MessageFolderImportPolicy {
-  ALL_FOLDERS = 'ALL_FOLDERS',
-  SELECTED_FOLDERS = 'SELECTED_FOLDERS',
-}
-
-export enum MessageChannelPendingGroupEmailsAction {
-  GROUP_EMAILS_DELETION = 'GROUP_EMAILS_DELETION',
-  GROUP_EMAILS_IMPORT = 'GROUP_EMAILS_IMPORT',
-  NONE = 'NONE',
-}
+export {
+  MessageChannelContactAutoCreationPolicy,
+  MessageChannelPendingGroupEmailsAction,
+  MessageChannelSyncStage,
+  MessageChannelSyncStatus,
+  MessageChannelType,
+  MessageChannelVisibility,
+  MessageFolderImportPolicy,
+};
 
 registerEnumType(MessageChannelVisibility, {
   name: 'MessageChannelVisibility',

packages/twenty-server/src/modules/messaging/common/standard-objects/message-folder.workspace-entity.ts

@@ -1,6 +1,9 @@
 import { registerEnumType } from '@nestjs/graphql';
 
-import { FieldMetadataType } from 'twenty-shared/types';
+import {
+  FieldMetadataType,
+  MessageFolderPendingSyncAction,
+} from 'twenty-shared/types';
 
 import { BaseWorkspaceEntity } from 'src/engine/twenty-orm/base.workspace-entity';
 import { type FieldTypeAndNameMetadata } from 'src/engine/workspace-manager/utils/get-ts-vector-column-expression.util';
@@ -8,10 +11,7 @@ import { type EntityRelation } from 'src/engine/workspace-manager/workspace-migr
 import { type MessageChannelMessageAssociationMessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association-message-folder.workspace-entity';
 import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 
-export enum MessageFolderPendingSyncAction {
-  FOLDER_DELETION = 'FOLDER_DELETION',
-  NONE = 'NONE',
-}
+export { MessageFolderPendingSyncAction };
 
 registerEnumType(MessageFolderPendingSyncAction, {
   name: 'MessageFolderPendingSyncAction',

packages/twenty-server/src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command.ts

@@ -3,10 +3,10 @@ import { Logger } from '@nestjs/common';
 import { Command, CommandRunner, Option } from 'nest-commander';
 import { isDefined } from 'twenty-shared/utils';
 
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { 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 = {
@@ -24,6 +24,7 @@ export class MessagingResetChannelCommand extends CommandRunner {
 
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService,
     private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
   ) {
@@ -39,21 +40,14 @@ export class MessagingResetChannelCommand extends CommandRunner {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          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 } : {}),
-        },
-      });
+      const messageChannels = await this.messageChannelDataAccessService.find(
+        workspaceId,
+        isDefined(messageChannelId) ? { id: messageChannelId } : {},
+      );
 
       if (messageChannels.length === 0) {
         this.logger.log(

packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts

@@ -3,6 +3,7 @@ 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 { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.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';
@@ -15,6 +16,7 @@ import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cl
     TypeOrmModule.forFeature([WorkspaceEntity]),
     DataSourceModule,
     MessagingCommonModule,
+    MessageChannelDataAccessModule,
   ],
   providers: [
     MessagingConnectedAccountDeletionCleanupJob,

packages/twenty-server/src/modules/messaging/message-folder-manager/messaging-folder-sync-manager.module.ts

@@ -4,6 +4,7 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module';
+import { MessageFolderDataAccessModule } from 'src/engine/metadata-modules/message-folder/data-access/message-folder-data-access.module';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { OAuth2ClientManagerModule } from 'src/modules/connected-account/oauth2-client-manager/oauth2-client-manager.module';
 import { GmailFoldersErrorHandlerService } from 'src/modules/messaging/message-folder-manager/drivers/gmail/services/gmail-folders-error-handler.service';
@@ -20,6 +21,7 @@ import { MessagingMicrosoftDriverModule } from 'src/modules/messaging/message-im
     FeatureFlagModule,
     WorkspaceDataSourceModule,
     DataSourceModule,
+    MessageFolderDataAccessModule,
     TypeOrmModule.forFeature([WorkspaceEntity]),
     OAuth2ClientManagerModule,
     MessagingGmailDriverModule,

packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.spec.ts

@@ -1,9 +1,11 @@
 import { Test, type TestingModule } from '@nestjs/testing';
 
 import { ConnectedAccountProvider } from 'twenty-shared/types';
+import { In } from 'typeorm';
 
 import { type DiscoveredMessageFolder } from 'src/modules/messaging/message-folder-manager/interfaces/message-folder-driver.interface';
 
+import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.service';
 import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
 import {
   MessageChannelContactAutoCreationPolicy,
@@ -83,39 +85,73 @@ const createMockExistingFolder = (
   ...overrides,
 });
 
+const getInValuesFromWhere = (externalIdClause: unknown): string[] => {
+  if (
+    externalIdClause &&
+    typeof externalIdClause === 'object' &&
+    '_value' in externalIdClause
+  ) {
+    return (externalIdClause as { _value: string[] })._value;
+  }
+
+  if (
+    externalIdClause &&
+    typeof externalIdClause === 'object' &&
+    'value' in externalIdClause
+  ) {
+    const value = (externalIdClause as { value: unknown }).value;
+
+    return Array.isArray(value) ? value : [];
+  }
+
+  return [];
+};
+
 describe('SyncMessageFoldersService', () => {
   let service: SyncMessageFoldersService;
   let gmailGetAllFoldersService: jest.Mocked<GmailGetAllFoldersService>;
 
-  let mockRepository: {
+  let mockMessageFolderDataAccessService: {
     delete: jest.Mock;
     update: jest.Mock;
-    updateMany: jest.Mock;
     save: jest.Mock;
+    find: jest.Mock;
   };
-  let mockTransactionManager: object;
+
+  let createdFolderRecords: Array<
+    Partial<MessageFolderWorkspaceEntity> & {
+      id: string;
+      externalId: string;
+    }
+  >;
 
   beforeEach(async () => {
-    mockRepository = {
+    createdFolderRecords = [];
+
+    mockMessageFolderDataAccessService = {
       delete: jest.fn(),
-      update: jest.fn(),
-      updateMany: jest.fn(),
-      save: jest.fn().mockImplementation((folders) =>
-        folders.map((folder: Partial<MessageFolderWorkspaceEntity>) => ({
+      update: jest.fn().mockResolvedValue(undefined),
+      save: jest.fn().mockImplementation(async (_workspaceId, folder) => {
+        createdFolderRecords.push({
           ...folder,
-          id: `new-folder-${Math.random().toString(36).substring(7)}`,
+          id: `new-folder-${createdFolderRecords.length}-${Math.random().toString(36).substring(7)}`,
           isSynced: false,
           syncCursor: null,
-        })),
-      ),
-    };
+          pendingSyncAction: MessageFolderPendingSyncAction.NONE,
+          externalId: folder.externalId as string,
+        });
+      }),
+      find: jest.fn().mockImplementation(async (_workspaceId, where) => {
+        if (!where?.externalId) {
+          return [];
+        }
 
-    mockTransactionManager = {};
+        const externalIds = getInValuesFromWhere(where.externalId);
 
-    const mockDataSource = {
-      transaction: jest
-        .fn()
-        .mockImplementation((callback) => callback(mockTransactionManager)),
+        return createdFolderRecords.filter((folder) =>
+          externalIds.includes(folder.externalId as string),
+        );
+      }),
     };
 
     const module: TestingModule = await Test.createTestingModule({
@@ -129,15 +165,15 @@ describe('SyncMessageFoldersService', () => {
               .mockImplementation((callback: () => any, _authContext?: any) =>
                 callback(),
               ),
-            getRepository: jest.fn().mockResolvedValue(mockRepository),
-            getDataSourceForWorkspace: jest
-              .fn()
-              .mockResolvedValue(mockDataSource),
-            getGlobalWorkspaceDataSource: jest
-              .fn()
-              .mockResolvedValue(mockDataSource),
+            getRepository: jest.fn(),
+            getDataSourceForWorkspace: jest.fn(),
+            getGlobalWorkspaceDataSource: jest.fn(),
           },
         },
+        {
+          provide: MessageFolderDataAccessService,
+          useValue: mockMessageFolderDataAccessService,
+        },
         {
           provide: GmailGetAllFoldersService,
           useValue: {
@@ -192,23 +228,23 @@ describe('SyncMessageFoldersService', () => {
           workspaceId,
         });
 
-        expect(mockRepository.save).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({
-              name: 'INBOX',
-              externalId: 'inbox-ext',
-              messageChannelId: 'channel-123',
-              isSentFolder: false,
-            }),
-            expect.objectContaining({
-              name: 'Sent',
-              externalId: 'sent-ext',
-              messageChannelId: 'channel-123',
-              isSentFolder: true,
-            }),
-          ]),
-          {},
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.save).toHaveBeenCalledWith(
+          workspaceId,
+          expect.objectContaining({
+            name: 'INBOX',
+            externalId: 'inbox-ext',
+            messageChannelId: 'channel-123',
+            isSentFolder: false,
+          }),
+        );
+        expect(mockMessageFolderDataAccessService.save).toHaveBeenCalledWith(
+          workspaceId,
+          expect.objectContaining({
+            name: 'Sent',
+            externalId: 'sent-ext',
+            messageChannelId: 'channel-123',
+            isSentFolder: true,
+          }),
         );
         expect(result).toHaveLength(2);
       });
@@ -239,15 +275,12 @@ describe('SyncMessageFoldersService', () => {
           workspaceId,
         });
 
-        expect(mockRepository.save).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({
-              name: 'Projects',
-              parentFolderId: 'parent-folder-id',
-            }),
-          ]),
-          {},
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.save).toHaveBeenCalledWith(
+          workspaceId,
+          expect.objectContaining({
+            name: 'Projects',
+            parentFolderId: 'parent-folder-id',
+          }),
         );
       });
     });
@@ -278,14 +311,10 @@ describe('SyncMessageFoldersService', () => {
           workspaceId,
         });
 
-        expect(mockRepository.updateMany).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({
-              criteria: 'folder-1',
-              partialEntity: expect.objectContaining({ name: 'Primary Inbox' }),
-            }),
-          ]),
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
+          workspaceId,
+          { id: 'folder-1' },
+          expect.objectContaining({ name: 'Primary Inbox' }),
         );
         expect(result).toContainEqual(
           expect.objectContaining({
@@ -322,16 +351,12 @@ describe('SyncMessageFoldersService', () => {
           workspaceId,
         });
 
-        expect(mockRepository.updateMany).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({
-              criteria: 'folder-1',
-              partialEntity: expect.objectContaining({
-                parentFolderId: 'new-parent-id',
-              }),
-            }),
-          ]),
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
+          workspaceId,
+          { id: 'folder-1' },
+          expect.objectContaining({
+            parentFolderId: 'new-parent-id',
+          }),
         );
       });
 
@@ -364,7 +389,9 @@ describe('SyncMessageFoldersService', () => {
           workspaceId,
         });
 
-        expect(mockRepository.updateMany).not.toHaveBeenCalled();
+        expect(
+          mockMessageFolderDataAccessService.update,
+        ).not.toHaveBeenCalled();
       });
     });
 
@@ -401,16 +428,12 @@ describe('SyncMessageFoldersService', () => {
           workspaceId,
         });
 
-        expect(mockRepository.updateMany).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({
-              criteria: 'folder-2',
-              partialEntity: expect.objectContaining({
-                pendingSyncAction: 'FOLDER_DELETION',
-              }),
-            }),
-          ]),
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
+          workspaceId,
+          { id: In(['folder-2']) },
+          expect.objectContaining({
+            pendingSyncAction: 'FOLDER_DELETION',
+          }),
         );
         expect(result).toContainEqual(
           expect.objectContaining({
@@ -468,32 +491,21 @@ describe('SyncMessageFoldersService', () => {
           workspaceId,
         });
 
-        expect(mockRepository.updateMany).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({
-              criteria: 'folder-to-delete',
-              partialEntity: expect.objectContaining({
-                pendingSyncAction: 'FOLDER_DELETION',
-              }),
-            }),
-          ]),
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
+          workspaceId,
+          { id: In(['folder-to-delete']) },
+          expect.objectContaining({
+            pendingSyncAction: 'FOLDER_DELETION',
+          }),
         );
-        expect(mockRepository.updateMany).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({
-              criteria: 'folder-to-update',
-              partialEntity: expect.objectContaining({ name: 'New Name' }),
-            }),
-          ]),
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
+          workspaceId,
+          { id: 'folder-to-update' },
+          expect.objectContaining({ name: 'New Name' }),
         );
-        expect(mockRepository.save).toHaveBeenCalledWith(
-          expect.arrayContaining([
-            expect.objectContaining({ externalId: 'new-ext' }),
-          ]),
-          {},
-          mockTransactionManager,
+        expect(mockMessageFolderDataAccessService.save).toHaveBeenCalledWith(
+          workspaceId,
+          expect.objectContaining({ externalId: 'new-ext' }),
         );
         expect(result).toHaveLength(4);
         expect(result).toContainEqual(

packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.ts

@@ -1,21 +1,20 @@
 import { Injectable } from '@nestjs/common';
 
 import { ConnectedAccountProvider } from 'twenty-shared/types';
+import { isDefined } from 'twenty-shared/utils';
+import { In } from 'typeorm';
 
 import {
   DiscoveredMessageFolder,
   MessageFolder,
 } from 'src/modules/messaging/message-folder-manager/interfaces/message-folder-driver.interface';
 
-import { WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
+import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.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 { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
 import { 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';
+import { MessageFolderPendingSyncAction } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
 import { GmailGetAllFoldersService } from 'src/modules/messaging/message-folder-manager/drivers/gmail/services/gmail-get-all-folders.service';
 import { ImapGetAllFoldersService } from 'src/modules/messaging/message-folder-manager/drivers/imap/services/imap-get-all-folders.service';
 import { MicrosoftGetAllFoldersService } from 'src/modules/messaging/message-folder-manager/drivers/microsoft/services/microsoft-get-all-folders.service';
@@ -28,6 +27,7 @@ import { computeUpdatedFolders } from 'src/modules/messaging/message-folder-mana
 export class SyncMessageFoldersService {
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageFolderDataAccessService: MessageFolderDataAccessService,
     private readonly gmailGetAllFoldersService: GmailGetAllFoldersService,
     private readonly microsoftGetAllFoldersService: MicrosoftGetAllFoldersService,
     private readonly imapGetAllFoldersService: ImapGetAllFoldersService,
@@ -132,60 +132,63 @@ export class SyncMessageFoldersService {
 
     const authContext = buildSystemAuthContext(workspaceId);
 
+    // TODO: Restore transaction wrapper once migration is complete — folder
+    // sync operations (create/update/delete) are no longer atomic since
+    // the data access layer routes writes across workspace and core schemas.
+    // Acceptable during transition as sync is idempotent and self-corrects.
     return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
       async () => {
-        const messageFolderRepository =
-          await this.globalWorkspaceOrmManager.getRepository<MessageFolderWorkspaceEntity>(
+        if (folderIdsToDelete.length > 0) {
+          await this.messageFolderDataAccessService.update(
             workspaceId,
-            'messageFolder',
+            { id: In(folderIdsToDelete) },
+            {
+              pendingSyncAction: MessageFolderPendingSyncAction.FOLDER_DELETION,
+            },
           );
+        }
 
-        const workspaceDataSource =
-          await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource();
+        if (foldersToUpdate.size > 0) {
+          for (const [id, data] of foldersToUpdate.entries()) {
+            await this.messageFolderDataAccessService.update(
+              workspaceId,
+              { id },
+              data,
+            );
+          }
+        }
 
-        return workspaceDataSource.transaction(
-          async (transactionManager: WorkspaceEntityManager) => {
-            if (folderIdsToDelete.length > 0) {
-              await messageFolderRepository.updateMany(
-                folderIdsToDelete.map((id) => ({
-                  criteria: id,
-                  partialEntity: {
-                    pendingSyncAction:
-                      MessageFolderPendingSyncAction.FOLDER_DELETION,
-                  },
-                })),
-                transactionManager,
-              );
-            }
+        if (foldersToCreate.length > 0) {
+          for (const folderToCreate of foldersToCreate) {
+            await this.messageFolderDataAccessService.save(
+              workspaceId,
+              folderToCreate,
+            );
+          }
+        }
 
-            if (foldersToUpdate.size > 0) {
-              await messageFolderRepository.updateMany(
-                Array.from(foldersToUpdate.entries()).map(([id, data]) => ({
-                  criteria: id,
-                  partialEntity: data,
-                })),
-                transactionManager,
-              );
-            }
+        const createdFolders =
+          foldersToCreate.length > 0
+            ? await this.messageFolderDataAccessService.find(workspaceId, {
+                messageChannelId,
+                externalId: In(
+                  foldersToCreate
+                    .map((folder) => folder.externalId)
+                    .filter(isDefined),
+                ),
+              })
+            : [];
 
-            const createdFolders =
-              foldersToCreate.length > 0
-                ? await messageFolderRepository.save(
-                    foldersToCreate,
-                    {},
-                    transactionManager,
-                  )
-                : [];
+        const updatedExistingFolders = computeUpdatedFolders({
+          existingFolders,
+          foldersToUpdate,
+          folderIdsToDelete,
+        });
 
-            const updatedExistingFolders = computeUpdatedFolders({
-              existingFolders,
-              foldersToUpdate,
-              folderIdsToDelete,
-            });
-
-            return [...updatedExistingFolders, ...createdFolders];
-          },
-        );
+        return [
+          ...updatedExistingFolders,
+          ...(createdFolders as MessageFolder[]),
+        ];
       },
       authContext,
     );

packages/twenty-server/src/modules/messaging/message-import-manager/commands/messaging-trigger-message-list-fetch.command.ts

@@ -5,16 +5,14 @@ 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 { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 {
   MessagingMessageListFetchJob,
   type MessagingMessageListFetchJobData,
 } from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job';
-import {
-  MessageChannelSyncStage,
-  type MessageChannelWorkspaceEntity,
-} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { MessageChannelSyncStage } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 
 type MessagingTriggerMessageListFetchCommandOptions = {
   workspaceId: string;
@@ -33,6 +31,7 @@ export class MessagingTriggerMessageListFetchCommand extends CommandRunner {
 
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     @InjectMessageQueue(MessageQueue.messagingQueue)
     private readonly messageQueueService: MessageQueueService,
   ) {
@@ -52,23 +51,14 @@ export class MessagingTriggerMessageListFetchCommand extends CommandRunner {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      const whereCondition: Record<string, unknown> = {
-        isSyncEnabled: true,
-        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
-      };
-
-      if (messageChannelId) {
-        whereCondition.id = messageChannelId;
-      }
-
-      const messageChannels =
-        await messageChannelRepository.find(whereCondition);
+      const messageChannels = await this.messageChannelDataAccessService.find(
+        workspaceId,
+        {
+          isSyncEnabled: true,
+          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+          ...(messageChannelId ? { id: messageChannelId } : {}),
+        },
+      );
 
       if (messageChannels.length === 0) {
         this.logger.warn(
@@ -83,10 +73,14 @@ export class MessagingTriggerMessageListFetchCommand extends CommandRunner {
       );
 
       for (const messageChannel of messageChannels) {
-        await messageChannelRepository.update(messageChannel.id, {
-          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
-          syncStageStartedAt: new Date().toISOString(),
-        });
+        await this.messageChannelDataAccessService.update(
+          workspaceId,
+          { id: messageChannel.id },
+          {
+            syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
+            syncStageStartedAt: new Date().toISOString(),
+          },
+        );
 
         await this.messageQueueService.add<MessagingMessageListFetchJobData>(
           MessagingMessageListFetchJob.name,

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts

@@ -3,6 +3,7 @@ import { Scope } from '@nestjs/common';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
@@ -18,6 +19,16 @@ import {
 import { MessagingMessageListFetchService } from 'src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service';
 import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service';
 
+const toIsoStringOrNull = (
+  value: string | Date | null | undefined,
+): string | null => {
+  if (value == null) {
+    return null;
+  }
+
+  return value instanceof Date ? value.toISOString() : value;
+};
+
 export type MessagingMessageListFetchJobData = {
   messageChannelId: string;
   workspaceId: string;
@@ -32,6 +43,7 @@ export class MessagingMessageListFetchJob {
     private readonly messagingMessageListFetchService: MessagingMessageListFetchService,
     private readonly messagingMonitoringService: MessagingMonitoringService,
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly messageImportErrorHandlerService: MessageImportExceptionHandlerService,
     private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
   ) {}
@@ -49,18 +61,15 @@ export class MessagingMessageListFetchJob {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      const messageChannel = await messageChannelRepository.findOne({
-        where: {
-          id: messageChannelId,
+      const messageChannel = await this.messageChannelDataAccessService.findOne(
+        workspaceId,
+        {
+          where: {
+            id: messageChannelId,
+          },
+          relations: ['connectedAccount', 'messageFolders'],
         },
-        relations: ['connectedAccount', 'messageFolders'],
-      });
+      );
 
       if (!messageChannel) {
         await this.messagingMonitoringService.track({
@@ -82,9 +91,9 @@ export class MessagingMessageListFetchJob {
       try {
         if (
           isThrottled(
-            messageChannel.syncStageStartedAt,
+            toIsoStringOrNull(messageChannel.syncStageStartedAt),
             messageChannel.throttleFailureCount,
-            messageChannel.throttleRetryAfter,
+            toIsoStringOrNull(messageChannel.throttleRetryAfter),
           )
         ) {
           await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
@@ -104,7 +113,7 @@ export class MessagingMessageListFetchJob {
         });
 
         await this.messagingMessageListFetchService.processMessageListFetch(
-          messageChannel,
+          messageChannel as unknown as MessageChannelWorkspaceEntity,
           workspaceId,
         );
 
@@ -118,7 +127,7 @@ export class MessagingMessageListFetchJob {
         await this.messageImportErrorHandlerService.handleDriverException(
           error,
           MessageImportSyncStep.MESSAGE_LIST_FETCH,
-          messageChannel,
+          messageChannel as unknown as MessageChannelWorkspaceEntity,
           workspaceId,
         );
       }

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job.ts

@@ -3,6 +3,7 @@ import { Scope } from '@nestjs/common';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
@@ -14,6 +15,16 @@ import {
 import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service';
 import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service';
 
+const toIsoStringOrNull = (
+  value: string | Date | null | undefined,
+): string | null => {
+  if (value == null) {
+    return null;
+  }
+
+  return value instanceof Date ? value.toISOString() : value;
+};
+
 export type MessagingMessagesImportJobData = {
   messageChannelId: string;
   workspaceId: string;
@@ -29,6 +40,7 @@ export class MessagingMessagesImportJob {
     private readonly messagingMonitoringService: MessagingMonitoringService,
     private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
   ) {}
 
   @Process(MessagingMessagesImportJob.name)
@@ -44,18 +56,15 @@ export class MessagingMessagesImportJob {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      const messageChannel = await messageChannelRepository.findOne({
-        where: {
-          id: messageChannelId,
+      const messageChannel = await this.messageChannelDataAccessService.findOne(
+        workspaceId,
+        {
+          where: {
+            id: messageChannelId,
+          },
+          relations: ['connectedAccount', 'messageFolders'],
         },
-        relations: ['connectedAccount', 'messageFolders'],
-      });
+      );
 
       if (!messageChannel) {
         await this.messagingMonitoringService.track({
@@ -80,9 +89,9 @@ export class MessagingMessagesImportJob {
 
       if (
         isThrottled(
-          messageChannel.syncStageStartedAt,
+          toIsoStringOrNull(messageChannel.syncStageStartedAt),
           messageChannel.throttleFailureCount,
-          messageChannel.throttleRetryAfter,
+          toIsoStringOrNull(messageChannel.throttleRetryAfter),
         )
       ) {
         await this.messageChannelSyncStatusService.markAsMessagesImportPending(
@@ -94,9 +103,12 @@ export class MessagingMessagesImportJob {
         return;
       }
 
+      const messageChannelWorkspace =
+        messageChannel as unknown as MessageChannelWorkspaceEntity;
+
       await this.messagingMessagesImportService.processMessageBatchImport(
-        messageChannel,
-        messageChannel.connectedAccount,
+        messageChannelWorkspace,
+        messageChannelWorkspace.connectedAccount,
         workspaceId,
       );
     }, authContext);

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts

@@ -5,15 +5,23 @@ import { In } from 'typeorm';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
-import {
-  MessageChannelSyncStage,
-  type MessageChannelWorkspaceEntity,
-} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { MessageChannelSyncStage } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import { isSyncStale } from 'src/modules/messaging/message-import-manager/utils/is-sync-stale.util';
 
+const toIsoStringOrNull = (
+  value: string | Date | null | undefined,
+): string | null => {
+  if (value == null) {
+    return null;
+  }
+
+  return value instanceof Date ? value.toISOString() : value;
+};
+
 export type MessagingOngoingStaleJobData = {
   workspaceId: string;
 };
@@ -26,6 +34,7 @@ export class MessagingOngoingStaleJob {
   private readonly logger = new Logger(MessagingOngoingStaleJob.name);
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
   ) {}
 
@@ -36,14 +45,9 @@ export class MessagingOngoingStaleJob {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      const messageChannels = await messageChannelRepository.find({
-        where: {
+      const messageChannels = await this.messageChannelDataAccessService.find(
+        workspaceId,
+        {
           syncStage: In([
             MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING,
             MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
@@ -51,10 +55,10 @@ export class MessagingOngoingStaleJob {
             MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
           ]),
         },
-      });
+      );
 
       for (const messageChannel of messageChannels) {
-        if (isSyncStale(messageChannel.syncStageStartedAt)) {
+        if (isSyncStale(toIsoStringOrNull(messageChannel.syncStageStartedAt))) {
           await this.messageChannelSyncStatusService.resetSyncStageStartedAt(
             [messageChannel.id],
             workspaceId,

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-relaunch-failed-message-channel.job.ts

@@ -3,12 +3,12 @@ import { Scope } from '@nestjs/common';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 {
   MessageChannelSyncStage,
   MessageChannelSyncStatus,
-  MessageChannelWorkspaceEntity,
 } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 
 export type MessagingRelaunchFailedMessageChannelJobData = {
@@ -23,6 +23,7 @@ export type MessagingRelaunchFailedMessageChannelJobData = {
 export class MessagingRelaunchFailedMessageChannelJob {
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
   ) {}
 
   @Process(MessagingRelaunchFailedMessageChannelJob.name)
@@ -32,23 +33,14 @@ export class MessagingRelaunchFailedMessageChannelJob {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-          { shouldBypassPermissionChecks: true },
-        );
-
-      const messageChannel = await messageChannelRepository.findOne({
-        where: {
-          id: messageChannelId,
-        },
-        relations: {
-          connectedAccount: {
-            accountOwner: true,
+      const messageChannel = await this.messageChannelDataAccessService.findOne(
+        workspaceId,
+        {
+          where: {
+            id: messageChannelId,
           },
         },
-      });
+      );
 
       if (
         !messageChannel ||
@@ -58,10 +50,14 @@ export class MessagingRelaunchFailedMessageChannelJob {
         return;
       }
 
-      await messageChannelRepository.update(messageChannelId, {
-        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
-        syncStatus: MessageChannelSyncStatus.ACTIVE,
-      });
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: messageChannelId },
+        {
+          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+          syncStatus: MessageChannelSyncStatus.ACTIVE,
+        },
+      );
     }, authContext);
   }
 }

packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts

@@ -4,6 +4,8 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-source.entity';
+import { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module';
+import { MessageFolderDataAccessModule } from 'src/engine/metadata-modules/message-folder/data-access/message-folder-data-access.module';
 import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { WorkspaceEventEmitterModule } from 'src/engine/workspace-event-emitter/workspace-event-emitter.module';
@@ -61,6 +63,8 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
     MessagingIMAPDriverModule,
     MessagingSmtpDriverModule,
     MessagingCommonModule,
+    MessageChannelDataAccessModule,
+    MessageFolderDataAccessModule,
     TypeOrmModule.forFeature([
       WorkspaceEntity,
       DataSourceEntity,

packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-message-list-fetch.service.spec.ts

@@ -24,6 +24,7 @@ import { MessagingMessageListFetchService } from 'src/modules/messaging/message-
 import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service';
 import { MessagingProcessFolderActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-folder-actions.service';
 import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
 
 describe('MessagingMessageListFetchService', () => {
   let messagingMessageListFetchService: MessagingMessageListFetchService;
@@ -229,6 +230,12 @@ describe('MessagingMessageListFetchService', () => {
             }),
           },
         },
+        {
+          provide: MessageChannelDataAccessService,
+          useValue: {
+            findOne: jest.fn().mockResolvedValue(undefined),
+          },
+        },
         {
           provide: MessagingCursorService,
           useValue: {

packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts

@@ -22,6 +22,7 @@ import { MessagingGetMessagesService } from 'src/modules/messaging/message-impor
 import { MessageImportExceptionHandlerService } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service';
 import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service';
 import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
 import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service';
 
 describe('MessagingMessagesImportService', () => {
@@ -127,6 +128,12 @@ describe('MessagingMessagesImportService', () => {
             .mockImplementation((fn: () => any, _authContext?: any) => fn()),
         },
       },
+      {
+        provide: MessageChannelDataAccessService,
+        useValue: {
+          update: jest.fn().mockResolvedValue(undefined),
+        },
+      },
       {
         provide: MessagingGetMessagesService,
         useValue: {

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-cursor.service.ts

@@ -1,14 +1,17 @@
 import { Injectable } from '@nestjs/common';
 
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
+import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.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 { 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';
 
 @Injectable()
 export class MessagingCursorService {
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
+    private readonly messageFolderDataAccessService: MessageFolderDataAccessService,
   ) {}
 
   public async updateCursor(
@@ -20,19 +23,9 @@ export class MessagingCursorService {
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-      const folderRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageFolderWorkspaceEntity>(
-          workspaceId,
-          'messageFolder',
-        );
-
       if (!folderId) {
-        await messageChannelRepository.update(
+        await this.messageChannelDataAccessService.update(
+          workspaceId,
           {
             id: messageChannel.id,
           },
@@ -48,7 +41,8 @@ export class MessagingCursorService {
           },
         );
       } else {
-        await folderRepository.update(
+        await this.messageFolderDataAccessService.update(
+          workspaceId,
           {
             id: folderId,
           },
@@ -56,7 +50,8 @@ export class MessagingCursorService {
             syncCursor: nextSyncCursor,
           },
         );
-        await messageChannelRepository.update(
+        await this.messageChannelDataAccessService.update(
+          workspaceId,
           {
             id: messageChannel.id,
           },

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts

@@ -7,8 +7,7 @@ import {
   type TwentyORMException,
   TwentyORMExceptionCode,
 } from 'src/engine/twenty-orm/exceptions/twenty-orm.exception';
-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 { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
 import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import {
   MessageChannelSyncStatus,
@@ -30,7 +29,7 @@ export enum MessageImportSyncStep {
 @Injectable()
 export class MessageImportExceptionHandlerService {
   constructor(
-    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
     private readonly exceptionHandlerService: ExceptionHandlerService,
   ) {}
@@ -152,37 +151,27 @@ export class MessageImportExceptionHandlerService {
       return;
     }
 
-    const authContext = buildSystemAuthContext(workspaceId);
+    await this.messageChannelDataAccessService.increment(
+      workspaceId,
+      { id: messageChannel.id },
+      'throttleFailureCount',
+      1,
+    );
 
-    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
+    const throttleRetryAfter =
+      exception instanceof MessageImportDriverException
+        ? exception.throttleRetryAfter
+        : undefined;
 
-      await messageChannelRepository.increment(
-        { id: messageChannel.id },
-        'throttleFailureCount',
-        1,
-        undefined,
-        ['throttleFailureCount', 'id'],
-      );
-
-      const throttleRetryAfter =
-        exception instanceof MessageImportDriverException
-          ? exception.throttleRetryAfter
-          : undefined;
-
-      await messageChannelRepository.update(
-        { id: messageChannel.id },
-        {
-          throttleRetryAfter: isDefined(throttleRetryAfter)
-            ? throttleRetryAfter.toISOString()
-            : null,
-        },
-      );
-    }, authContext);
+    await this.messageChannelDataAccessService.update(
+      workspaceId,
+      { id: messageChannel.id },
+      {
+        throttleRetryAfter: isDefined(throttleRetryAfter)
+          ? throttleRetryAfter.toISOString()
+          : null,
+      },
+    );
 
     switch (syncStep) {
       case MessageImportSyncStep.MESSAGE_LIST_FETCH:

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts

@@ -8,6 +8,7 @@ import { In, MoreThanOrEqual } 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 { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
@@ -41,6 +42,7 @@ export class MessagingMessageListFetchService {
     private readonly cacheStorage: CacheStorageService,
     private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly messagingGetMessageListService: MessagingGetMessageListService,
     private readonly messageImportErrorHandlerService: MessageImportExceptionHandlerService,
     private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
@@ -78,15 +80,9 @@ export class MessagingMessageListFetchService {
           `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Processing message list fetch`,
         );
 
-        const messageChannelRepository =
-          await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-            workspaceId,
-            'messageChannel',
-          );
-
         const freshMessageChannel =
           pendingGroupEmailActionsProcessed || pendingFolderActionsProcessed
-            ? await messageChannelRepository.findOne({
+            ? await this.messageChannelDataAccessService.findOne(workspaceId, {
                 where: {
                   id: messageChannel.id,
                 },

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts

@@ -5,6 +5,7 @@ import { isDefined } from 'twenty-shared/utils';
 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 { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
 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';
@@ -46,6 +47,7 @@ export class MessagingMessagesImportService {
     private readonly blocklistRepository: BlocklistRepository,
     private readonly emailAliasManagerService: EmailAliasManagerService,
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly messagingGetMessagesService: MessagingGetMessagesService,
     private readonly messageImportErrorHandlerService: MessageImportExceptionHandlerService,
     private readonly messagingAccountAuthenticationService: MessagingAccountAuthenticationService,
@@ -196,13 +198,8 @@ export class MessagingMessagesImportService {
           );
         }
 
-        const messageChannelRepository =
-          await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-            workspaceId,
-            'messageChannel',
-          );
-
-        await messageChannelRepository.update(
+        await this.messageChannelDataAccessService.update(
+          workspaceId,
           {
             id: messageChannel.id,
           },

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-folder-actions.service.ts

@@ -3,7 +3,7 @@ import { Injectable, Logger } from '@nestjs/common';
 import { isDefined } from 'twenty-shared/utils';
 import { In } from 'typeorm';
 
-import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
+import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.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 { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
@@ -21,6 +21,7 @@ export class MessagingProcessFolderActionsService {
 
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageFolderDataAccessService: MessageFolderDataAccessService,
     private readonly messagingDeleteFolderMessagesService: MessagingDeleteFolderMessagesService,
   ) {}
 
@@ -91,41 +92,27 @@ export class MessagingProcessFolderActionsService {
 
       await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
         async () => {
-          const workspaceDataSource =
-            await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource();
+          if (processedFolderIds.length > 0) {
+            await this.messageFolderDataAccessService.update(
+              workspaceId,
+              { id: In(processedFolderIds) },
+              { pendingSyncAction: MessageFolderPendingSyncAction.NONE },
+            );
 
-          await workspaceDataSource?.transaction(
-            async (transactionManager: WorkspaceEntityManager) => {
-              const messageFolderRepository =
-                await this.globalWorkspaceOrmManager.getRepository<MessageFolderWorkspaceEntity>(
-                  workspaceId,
-                  'messageFolder',
-                );
+            this.logger.debug(
+              `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Reset pendingSyncAction to NONE for ${processedFolderIds.length} folders`,
+            );
+          }
 
-              if (processedFolderIds.length > 0) {
-                await messageFolderRepository.update(
-                  { id: In(processedFolderIds) },
-                  { pendingSyncAction: MessageFolderPendingSyncAction.NONE },
-                  transactionManager,
-                );
+          if (folderIdsToDelete.length > 0) {
+            await this.messageFolderDataAccessService.delete(workspaceId, {
+              id: In(folderIdsToDelete),
+            });
 
-                this.logger.debug(
-                  `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Reset pendingSyncAction to NONE for ${processedFolderIds.length} folders`,
-                );
-              }
-
-              if (folderIdsToDelete.length > 0) {
-                await messageFolderRepository.delete(
-                  { id: In(folderIdsToDelete) },
-                  transactionManager,
-                );
-
-                this.logger.log(
-                  `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Deleted ${folderIdsToDelete.length} folders`,
-                );
-              }
-            },
-          );
+            this.logger.log(
+              `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Deleted ${folderIdsToDelete.length} folders`,
+            );
+          }
         },
         authContext,
       );

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts

@@ -2,6 +2,8 @@ import { Injectable, Logger } from '@nestjs/common';
 
 import { isDefined } from 'twenty-shared/utils';
 
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service';
+import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.service';
 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';
 import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
@@ -9,7 +11,6 @@ import {
   MessageChannelPendingGroupEmailsAction,
   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';
 import { MessagingDeleteGroupEmailMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-group-email-messages.service';
 
 @Injectable()
@@ -20,6 +21,8 @@ export class MessagingProcessGroupEmailActionsService {
 
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
+    private readonly messageFolderDataAccessService: MessageFolderDataAccessService,
     private readonly messagingDeleteGroupEmailMessagesService: MessagingDeleteGroupEmailMessagesService,
   ) {}
 
@@ -28,24 +31,15 @@ export class MessagingProcessGroupEmailActionsService {
     workspaceId: string,
     pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction,
   ): Promise<void> {
-    const authContext = buildSystemAuthContext(workspaceId);
+    await this.messageChannelDataAccessService.update(
+      workspaceId,
+      { id: messageChannel.id },
+      { pendingGroupEmailsAction },
+    );
 
-    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      const messageChannelRepository =
-        await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-        );
-
-      await messageChannelRepository.update(
-        { id: messageChannel.id },
-        { pendingGroupEmailsAction },
-      );
-
-      this.logger.debug(
-        `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Marked message channel as pending group emails action: ${pendingGroupEmailsAction}`,
-      );
-    }, authContext);
+    this.logger.debug(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Marked message channel as pending group emails action: ${pendingGroupEmailsAction}`,
+    );
   }
 
   async processGroupEmailActions(
@@ -74,12 +68,6 @@ export class MessagingProcessGroupEmailActionsService {
       await workspaceDataSource?.transaction(
         async (transactionManager: WorkspaceEntityManager) => {
           try {
-            const messageChannelRepository =
-              await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-                workspaceId,
-                'messageChannel',
-              );
-
             switch (pendingGroupEmailsAction) {
               case MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_DELETION:
                 await this.handleGroupEmailsDeletion(
@@ -97,7 +85,8 @@ export class MessagingProcessGroupEmailActionsService {
                 break;
             }
 
-            await messageChannelRepository.update(
+            await this.messageChannelDataAccessService.update(
+              workspaceId,
               { id: messageChannel.id },
               {
                 pendingGroupEmailsAction:
@@ -167,27 +156,17 @@ export class MessagingProcessGroupEmailActionsService {
     messageChannelId: string;
     transactionManager: WorkspaceEntityManager;
   }) {
-    const messageChannelRepository =
-      await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
-        workspaceId,
-        'messageChannel',
-      );
-
-    await messageChannelRepository.update(
-      messageChannelId,
+    await this.messageChannelDataAccessService.update(
+      workspaceId,
+      { id: messageChannelId },
       {
         syncCursor: '',
       },
       transactionManager,
     );
 
-    const messageFolderRepository =
-      await this.globalWorkspaceOrmManager.getRepository<MessageFolderWorkspaceEntity>(
-        workspaceId,
-        'messageFolder',
-      );
-
-    await messageFolderRepository.update(
+    await this.messageFolderDataAccessService.update(
+      workspaceId,
       { messageChannelId },
       { syncCursor: '' },
       transactionManager,

packages/twenty-server/src/modules/messaging/monitoring/crons/jobs/messaging-message-channel-sync-status-monitoring.cron.job.ts

@@ -10,9 +10,9 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.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 { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service';
 
 export const MESSAGING_MESSAGE_CHANNEL_SYNC_STATUS_MONITORING_CRON_PATTERN =
@@ -25,6 +25,7 @@ export class MessagingMessageChannelSyncStatusMonitoringCronJob {
     private readonly workspaceRepository: Repository<WorkspaceEntity>,
     private readonly messagingMonitoringService: MessagingMonitoringService,
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly messageChannelDataAccessService: MessageChannelDataAccessService,
     private readonly exceptionHandlerService: ExceptionHandlerService,
   ) {}
 
@@ -51,14 +52,13 @@ export class MessagingMessageChannelSyncStatusMonitoringCronJob {
 
         await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
           async () => {
-            const messageChannelRepository =
-              await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
+            const messageChannels =
+              await this.messageChannelDataAccessService.findMany(
                 activeWorkspace.id,
-                'messageChannel',
+                {
+                  select: ['id', 'syncStatus', 'connectedAccountId'],
+                },
               );
-            const messageChannels = await messageChannelRepository.find({
-              select: ['id', 'syncStatus', 'connectedAccountId'],
-            });
 
             for (const messageChannel of messageChannels) {
               if (!messageChannel.syncStatus) {

packages/twenty-server/src/modules/messaging/monitoring/messaging-monitoring.module.ts

@@ -4,6 +4,7 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { AuditModule } from 'src/engine/core-modules/audit/audit.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-source.entity';
+import { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module';
 import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module';
 import { MessagingMessageChannelSyncStatusMonitoringCronCommand } from 'src/modules/messaging/monitoring/crons/commands/messaging-message-channel-sync-status-monitoring.cron.command';
 import { MessagingMessageChannelSyncStatusMonitoringCronJob } from 'src/modules/messaging/monitoring/crons/jobs/messaging-message-channel-sync-status-monitoring.cron.job';
@@ -12,6 +13,7 @@ import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/ser
 @Module({
   imports: [
     AuditModule,
+    MessageChannelDataAccessModule,
     MessagingCommonModule,
     TypeOrmModule.forFeature([WorkspaceEntity]),
     TypeOrmModule.forFeature([DataSourceEntity]),