Refactor timeline activity insert (#13696)

In this PR: - refactor timelineActivity computation to batch it - make sure to pass the authContext to insert-query-builder in order to pass user information to timelineActivity - refactor message-participant /calendar-participant creation to batch more - fix favorite on view race condition (FE) - deprecate PARTIAL_CALENDAR_EVENT_FETCH_LIST syncStage as we will deprecate partial vs full notion (we will just leverage cursor emptyness or not) - introduce calendar / messging SCHEDULED syncStage that will allow better performance granularity later - activate quick message import after message list fetch to speed performance on small message lists
2025-08-06 22:11:26 +02:00
parent 453a6167a5
commit 5631dd122e
34 changed files with 684 additions and 572 deletions
@@ -54,7 +54,7 @@ export class BlocklistReimportCalendarEventsJob {
        },
      });

-      await this.calendarChannelSyncStatusService.resetAndScheduleFullCalendarEventListFetch(
+      await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
        calendarChannels.map((calendarChannel) => calendarChannel.id),
        workspaceId,
      );
@@ -57,20 +57,9 @@ export class CalendarEventListFetchJob {
    }

    switch (calendarChannel.syncStage) {
-      case CalendarChannelSyncStage.FULL_CALENDAR_EVENT_LIST_FETCH_PENDING:
-        await calendarChannelRepository.update(calendarChannelId, {
-          syncCursor: '',
-          syncStageStartedAt: null,
-        });
-
-        await this.calendarFetchEventsService.fetchCalendarEvents(
-          calendarChannel,
-          calendarChannel.connectedAccount,
-          workspaceId,
-        );
-        break;
-
-      case CalendarChannelSyncStage.PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING:
+      case CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED:
+      case CalendarChannelSyncStage.PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING: // DEPRECATED
+      case CalendarChannelSyncStage.FULL_CALENDAR_EVENT_LIST_FETCH_PENDING: // WILL BE DEPRECATED
        await this.calendarFetchEventsService.fetchCalendarEvents(
          calendarChannel,
          calendarChannel.connectedAccount,
@@ -54,7 +54,7 @@ export class CalendarOngoingStaleJob {

        switch (calendarChannel.syncStage) {
          case CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING:
-            await this.calendarChannelSyncStatusService.schedulePartialCalendarEventListFetch(
+            await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
              [calendarChannel.id],
            );
            break;
@@ -130,13 +130,13 @@ export class CalendarEventImportErrorHandlerService {

    switch (syncStep) {
      case CalendarEventImportSyncStep.FULL_CALENDAR_EVENT_LIST_FETCH:
-        await this.calendarChannelSyncStatusService.scheduleFullCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
          [calendarChannel.id],
        );
        break;

      case CalendarEventImportSyncStep.PARTIAL_CALENDAR_EVENT_LIST_FETCH:
-        await this.calendarChannelSyncStatusService.schedulePartialCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
          [calendarChannel.id],
        );
        break;
@@ -205,7 +205,7 @@ export class CalendarEventImportErrorHandlerService {
      return;
    }

-    await this.calendarChannelSyncStatusService.resetAndScheduleFullCalendarEventListFetch(
+    await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
      [calendarChannel.id],
      workspaceId,
    );
@@ -61,7 +61,7 @@ export class CalendarEventsImportService {
        );

        if (!eventIdsToFetch || eventIdsToFetch.length === 0) {
-          await this.calendarChannelSyncStatusService.markAsCompletedAndSchedulePartialCalendarEventListFetch(
+          await this.calendarChannelSyncStatusService.markAsCompletedAndScheduleCalendarEventListFetch(
            [calendarChannel.id],
          );

@@ -82,7 +82,7 @@ export class CalendarEventsImportService {
      }

      if (!calendarEvents || calendarEvents?.length === 0) {
-        await this.calendarChannelSyncStatusService.schedulePartialCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
          [calendarChannel.id],
        );
      }
@@ -134,7 +134,7 @@ export class CalendarEventsImportService {
        workspaceId,
      );

-      await this.calendarChannelSyncStatusService.markAsCompletedAndSchedulePartialCalendarEventListFetch(
+      await this.calendarChannelSyncStatusService.markAsCompletedAndScheduleCalendarEventListFetch(
        [calendarChannel.id],
      );
    } catch (error) {
@@ -79,7 +79,7 @@ export class CalendarFetchEventsService {
          },
        );

-        await this.calendarChannelSyncStatusService.schedulePartialCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
          [calendarChannel.id],
        );
      }
@@ -1,12 +1,12 @@
 import { Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';

+import { isDefined } from 'twenty-shared/utils';
 import { Repository } from 'typeorm';

 import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
 import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
-import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util';
 import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
 import { CalendarEventParticipantWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-event-participant.workspace-entity';
 import { TimelineActivityRepository } from 'src/modules/timeline/repositories/timeline-activity.repository';
@@ -28,48 +28,50 @@ export class CalendarEventParticipantListener {
      participants: CalendarEventParticipantWorkspaceEntity[];
    }>,
  ): Promise<void> {
-    const workspaceId = batchEvent.workspaceId;
+    const calendarEventObjectMetadata =
+      await this.objectMetadataRepository.findOneOrFail({
+        where: {
+          nameSingular: 'calendarEvent',
+          workspaceId: batchEvent.workspaceId,
+        },
+      });

-    // TODO: Refactor to insertTimelineActivitiesForObject once
-    for (const eventPayload of batchEvent.events) {
-      const calendarEventParticipants = eventPayload.participants;
-      const workspaceMemberId = eventPayload.workspaceMemberId;
-
-      // TODO: move to a job?
-
-      const dataSourceSchema = getWorkspaceSchemaName(workspaceId);
-
-      const calendarEventObjectMetadata =
-        await this.objectMetadataRepository.findOneOrFail({
-          where: {
-            nameSingular: 'calendarEvent',
-            workspaceId,
-          },
-        });
+    const timelineActivityPayloads = batchEvent.events.flatMap((event) => {
+      const calendarEventParticipants = event.participants ?? [];

      const calendarEventParticipantsWithPersonId =
-        calendarEventParticipants.filter((participant) => participant.personId);
+        calendarEventParticipants.filter((participant) =>
+          isDefined(participant.personId),
+        );

      if (calendarEventParticipantsWithPersonId.length === 0) {
-        continue;
+        return;
      }

-      await this.timelineActivityRepository.insertTimelineActivitiesForObject(
-        'person',
-        calendarEventParticipantsWithPersonId.map((participant) => ({
-          dataSourceSchema,
-          name: 'calendarEvent.linked',
-          properties: null,
-          objectName: 'calendarEvent',
-          recordId: participant.personId,
-          workspaceMemberId,
-          workspaceId,
-          linkedObjectMetadataId: calendarEventObjectMetadata.id,
-          linkedRecordId: participant.calendarEventId,
-          linkedRecordCachedName: '',
-        })),
-        workspaceId,
-      );
-    }
+      return calendarEventParticipantsWithPersonId
+        .map((participant) => {
+          if (!isDefined(participant.personId)) {
+            return;
+          }
+
+          return {
+            name: 'message.linked',
+            properties: {},
+            objectSingularName: 'person',
+            recordId: participant.personId,
+            workspaceMemberId: event.workspaceMemberId,
+            linkedObjectMetadataId: calendarEventObjectMetadata.id,
+            linkedRecordId: participant.calendarEventId,
+            linkedRecordCachedName: '',
+          };
+        })
+        .filter(isDefined);
+    });
+
+    await this.timelineActivityRepository.upsertTimelineActivities({
+      objectSingularName: 'person',
+      workspaceId: batchEvent.workspaceId,
+      payloads: timelineActivityPayloads.filter(isDefined),
+    });
  }
 }
@@ -27,9 +27,7 @@ export class CalendarChannelSyncStatusService {
    private readonly metricsService: MetricsService,
  ) {}

-  public async scheduleFullCalendarEventListFetch(
-    calendarChannelIds: string[],
-  ) {
+  public async scheduleCalendarEventListFetch(calendarChannelIds: string[]) {
    if (!calendarChannelIds.length) {
      return;
    }
@@ -45,24 +43,6 @@ export class CalendarChannelSyncStatusService {
    });
  }

-  public async schedulePartialCalendarEventListFetch(
-    calendarChannelIds: string[],
-  ) {
-    if (!calendarChannelIds.length) {
-      return;
-    }
-
-    const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
-        'calendarChannel',
-      );
-
-    await calendarChannelRepository.update(calendarChannelIds, {
-      syncStage:
-        CalendarChannelSyncStage.PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING,
-    });
-  }
-
  public async markAsCalendarEventListFetchOngoing(
    calendarChannelIds: string[],
  ) {
@@ -82,7 +62,7 @@ export class CalendarChannelSyncStatusService {
    });
  }

-  public async resetAndScheduleFullCalendarEventListFetch(
+  public async resetAndScheduleCalendarEventListFetch(
    calendarChannelIds: string[],
    workspaceId: string,
  ) {
@@ -107,7 +87,7 @@ export class CalendarChannelSyncStatusService {
      throttleFailureCount: 0,
    });

-    await this.scheduleFullCalendarEventListFetch(calendarChannelIds);
+    await this.scheduleCalendarEventListFetch(calendarChannelIds);
  }

  public async resetSyncStageStartedAt(calendarChannelIds: string[]) {
@@ -156,7 +136,7 @@ export class CalendarChannelSyncStatusService {
    });
  }

-  public async markAsCompletedAndSchedulePartialCalendarEventListFetch(
+  public async markAsCompletedAndScheduleCalendarEventListFetch(
    calendarChannelIds: string[],
  ) {
    if (!calendarChannelIds.length) {
@@ -177,7 +157,7 @@ export class CalendarChannelSyncStatusService {
      syncedAt: new Date().toISOString(),
    });

-    await this.schedulePartialCalendarEventListFetch(calendarChannelIds);
+    await this.scheduleCalendarEventListFetch(calendarChannelIds);

    await this.metricsService.batchIncrementCounter({
      key: MetricsKeys.CalendarEventSyncJobActive,
@@ -35,10 +35,13 @@ export enum CalendarChannelSyncStatus {
 }

 export enum CalendarChannelSyncStage {
-  FULL_CALENDAR_EVENT_LIST_FETCH_PENDING = 'FULL_CALENDAR_EVENT_LIST_FETCH_PENDING',
-  PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING = 'PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING',
+  FULL_CALENDAR_EVENT_LIST_FETCH_PENDING = 'FULL_CALENDAR_EVENT_LIST_FETCH_PENDING', // WILL BE DEPRECATED
+  PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING = 'PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING', // DEPRECATED
+  CALENDAR_EVENT_LIST_FETCH_PENDING = 'CALENDAR_EVENT_LIST_FETCH_PENDING',
+  CALENDAR_EVENT_LIST_FETCH_SCHEDULED = 'CALENDAR_EVENT_LIST_FETCH_SCHEDULED',
  CALENDAR_EVENT_LIST_FETCH_ONGOING = 'CALENDAR_EVENT_LIST_FETCH_ONGOING',
  CALENDAR_EVENTS_IMPORT_PENDING = 'CALENDAR_EVENTS_IMPORT_PENDING',
+  CALENDAR_EVENTS_IMPORT_SCHEDULED = 'CALENDAR_EVENTS_IMPORT_SCHEDULED',
  CALENDAR_EVENTS_IMPORT_ONGOING = 'CALENDAR_EVENTS_IMPORT_ONGOING',
  FAILED = 'FAILED',
 }
@@ -137,17 +140,16 @@ export class CalendarChannelWorkspaceEntity extends BaseWorkspaceEntity {
    icon: 'IconStatusChange',
    options: [
      {
-        value: CalendarChannelSyncStage.FULL_CALENDAR_EVENT_LIST_FETCH_PENDING,
-        label: 'Full calendar event list fetch pending',
+        value: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
+        label: 'Calendar event list fetch pending',
        position: 0,
        color: 'blue',
      },
      {
-        value:
-          CalendarChannelSyncStage.PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING,
-        label: 'Partial calendar event list fetch pending',
+        value: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED,
+        label: 'Calendar event list fetch scheduled',
        position: 1,
-        color: 'blue',
+        color: 'green',
      },
      {
        value: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING,
@@ -161,18 +163,37 @@ export class CalendarChannelWorkspaceEntity extends BaseWorkspaceEntity {
        position: 3,
        color: 'blue',
      },
+      {
+        value: CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED,
+        label: 'Calendar events import scheduled',
+        position: 4,
+        color: 'green',
+      },
      {
        value: CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_ONGOING,
        label: 'Calendar events import ongoing',
-        position: 4,
+        position: 5,
        color: 'orange',
      },
      {
        value: CalendarChannelSyncStage.FAILED,
        label: 'Failed',
-        position: 5,
+        position: 6,
        color: 'red',
      },
+      {
+        value: CalendarChannelSyncStage.FULL_CALENDAR_EVENT_LIST_FETCH_PENDING,
+        label: 'Full calendar event list fetch pending',
+        position: 7,
+        color: 'blue',
+      },
+      {
+        value:
+          CalendarChannelSyncStage.PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING,
+        label: 'Partial calendar event list fetch pending',
+        position: 8,
+        color: 'blue',
+      },
    ],
    defaultValue: `'${CalendarChannelSyncStage.FULL_CALENDAR_EVENT_LIST_FETCH_PENDING}'`,
  })
@@ -159,7 +159,7 @@ export class MessageChannelSyncStatusService {
    });
  }

-  public async markAsFailedAndFlushMessagesToImport(
+  public async markAsFailed(
    messageChannelIds: string[],
    workspaceId: string,
    syncStatus:
@@ -170,12 +170,6 @@ export class MessageChannelSyncStatusService {
      return;
    }

-    for (const messageChannelId of messageChannelIds) {
-      await this.cacheStorage.del(
-        `messages-to-import:${workspaceId}:${messageChannelId}`,
-      );
-    }
-
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
@@ -31,10 +31,13 @@ export enum MessageChannelSyncStatus {
 }

 export enum MessageChannelSyncStage {
-  FULL_MESSAGE_LIST_FETCH_PENDING = 'FULL_MESSAGE_LIST_FETCH_PENDING', // TODO: rename to MESSAGE_LIST_FETCH_PENDING
-  PARTIAL_MESSAGE_LIST_FETCH_PENDING = 'PARTIAL_MESSAGE_LIST_FETCH_PENDING', // TODO: to be removed, deprecated
+  FULL_MESSAGE_LIST_FETCH_PENDING = 'FULL_MESSAGE_LIST_FETCH_PENDING', // WILL BE DEPRECATED
+  PARTIAL_MESSAGE_LIST_FETCH_PENDING = 'PARTIAL_MESSAGE_LIST_FETCH_PENDING', // DEPRECATED
+  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',
 }
@@ -291,16 +294,16 @@ export class MessageChannelWorkspaceEntity extends BaseWorkspaceEntity {
    icon: 'IconStatusChange',
    options: [
      {
-        value: MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING, // TODO: Rename to MESSAGE_LIST_FETCH_PENDING
-        label: 'Full messages list fetch pending',
+        value: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+        label: 'Messages list fetch pending',
        position: 0,
        color: 'blue',
      },
      {
-        value: MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING, // TODO: Deprecate
-        label: 'Partial messages list fetch pending',
+        value: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
+        label: 'Messages list fetch scheduled',
        position: 1,
-        color: 'blue',
+        color: 'green',
      },
      {
        value: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
@@ -314,18 +317,36 @@ export class MessageChannelWorkspaceEntity extends BaseWorkspaceEntity {
        position: 3,
        color: 'blue',
      },
+      {
+        value: MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED,
+        label: 'Messages import scheduled',
+        position: 4,
+        color: 'green',
+      },
      {
        value: MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING,
        label: 'Messages import ongoing',
-        position: 4,
+        position: 5,
        color: 'orange',
      },
      {
        value: MessageChannelSyncStage.FAILED,
        label: 'Failed',
-        position: 5,
+        position: 6,
        color: 'red',
      },
+      {
+        value: MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING, // WILL BE DEPRECATED
+        label: 'Full messages list fetch pending',
+        position: 7,
+        color: 'blue',
+      },
+      {
+        value: MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING, // DEPRECATED
+        label: 'Partial messages list fetch pending',
+        position: 8,
+        color: 'blue',
+      },
    ],
    defaultValue: `'${MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING}'`,
  })
@@ -85,8 +85,9 @@ export class MessagingMessageListFetchJob {
      );

      switch (messageChannel.syncStage) {
-        case MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING: // TODO: deprecate as we introduce MESSAGE_LIST_FETCH_PENDING
-        case MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING:
+        case MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING:
+        case MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING: // DEPRECATED
+        case MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING: // WILL BE DEPRECATED
          await this.messagingMonitoringService.track({
            eventName: 'full_message_list_fetch.started',
            workspaceId,
@@ -112,7 +112,7 @@ export class MessageImportExceptionHandlerService {
    if (
      messageChannel.throttleFailureCount >= MESSAGING_THROTTLE_MAX_ATTEMPTS
    ) {
-      await this.messageChannelSyncStatusService.markAsFailedAndFlushMessagesToImport(
+      await this.messageChannelSyncStatusService.markAsFailed(
        [messageChannel.id],
        workspaceId,
        MessageChannelSyncStatus.FAILED_UNKNOWN,
@@ -172,7 +172,7 @@ export class MessageImportExceptionHandlerService {
    messageChannel: Pick<MessageChannelWorkspaceEntity, 'id'>,
    workspaceId: string,
  ): Promise<void> {
-    await this.messageChannelSyncStatusService.markAsFailedAndFlushMessagesToImport(
+    await this.messageChannelSyncStatusService.markAsFailed(
      [messageChannel.id],
      workspaceId,
      MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
@@ -184,7 +184,7 @@ export class MessageImportExceptionHandlerService {
    messageChannel: Pick<MessageChannelWorkspaceEntity, 'id'>,
    workspaceId: string,
  ): Promise<void> {
-    await this.messageChannelSyncStatusService.markAsFailedAndFlushMessagesToImport(
+    await this.messageChannelSyncStatusService.markAsFailed(
      [messageChannel.id],
      workspaceId,
      MessageChannelSyncStatus.FAILED_UNKNOWN,
@@ -213,7 +213,7 @@ export class MessageImportExceptionHandlerService {
    messageChannel: Pick<MessageChannelWorkspaceEntity, 'id'>,
    workspaceId: string,
  ): Promise<void> {
-    await this.messageChannelSyncStatusService.markAsFailedAndFlushMessagesToImport(
+    await this.messageChannelSyncStatusService.markAsFailed(
      [messageChannel.id],
      workspaceId,
      MessageChannelSyncStatus.FAILED_UNKNOWN,
@@ -13,6 +13,7 @@ import { MessagingCursorService } from 'src/modules/messaging/message-import-man
 import { MessagingGetMessageListService } from 'src/modules/messaging/message-import-manager/services/messaging-get-message-list.service';
 import { MessageImportExceptionHandlerService } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service';
 import { MessagingMessageListFetchService } from 'src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service';
+import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service';

 describe('MessagingMessageListFetchService', () => {
  let messagingMessageListFetchService: MessagingMessageListFetchService;
@@ -77,6 +78,7 @@ describe('MessagingMessageListFetchService', () => {
          provide: CacheStorageNamespace.ModuleMessaging,
          useValue: {
            setAdd: jest.fn().mockResolvedValue(undefined),
+            del: jest.fn().mockResolvedValue(undefined),
          },
        },
        {
@@ -119,6 +121,12 @@ describe('MessagingMessageListFetchService', () => {
              }),
          },
        },
+        {
+          provide: MessagingMessagesImportService,
+          useValue: {
+            processMessageBatchImport: jest.fn().mockResolvedValue(undefined),
+          },
+        },
        {
          provide: MessageChannelSyncStatusService,
          useValue: {
@@ -148,6 +156,7 @@ describe('MessagingMessageListFetchService', () => {
          provide: CacheStorageService,
          useValue: {
            setAdd: jest.fn().mockResolvedValue(undefined),
+            del: jest.fn().mockResolvedValue(undefined),
          },
        },
        {
@@ -169,6 +178,7 @@ describe('MessagingMessageListFetchService', () => {
      module.get<MessagingMessageListFetchService>(
        MessagingMessageListFetchService,
      );
+
    messagingGetMessageListService = module.get<MessagingGetMessageListService>(
      MessagingGetMessageListService,
    );
@@ -8,7 +8,10 @@ import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/typ
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
-import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import {
+  MessageChannelSyncStage,
+  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';
 import { MessagingCursorService } from 'src/modules/messaging/message-import-manager/services/messaging-cursor.service';
 import { MessagingGetMessageListService } from 'src/modules/messaging/message-import-manager/services/messaging-get-message-list.service';
@@ -16,6 +19,11 @@ import {
  MessageImportExceptionHandlerService,
  MessageImportSyncStep,
 } 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';
+
+const MAX_MESSAGE_COUNT_FOR_QUICK_IMPORT = 100;
+const ONE_WEEK_IN_MILLISECONDS = 7 * 24 * 60 * 60 * 1000;
+
@Injectable()
 export class MessagingMessageListFetchService {
  constructor(
@@ -27,6 +35,7 @@ export class MessagingMessageListFetchService {
    private readonly messageImportErrorHandlerService: MessageImportExceptionHandlerService,
    private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
    private readonly messagingCursorService: MessagingCursorService,
+    private readonly messagingMessagesImportService: MessagingMessagesImportService,
  ) {}

  public async processMessageListFetch(
@@ -43,6 +52,15 @@ export class MessagingMessageListFetchService {
          messageChannel,
        );

+      await this.cacheStorage.del(
+        `messages-to-import:${workspaceId}:${messageChannel.id}`,
+      );
+
+      const totalMessageCount = messageLists.reduce(
+        (acc, messageList) => acc + messageList.messageExternalIds.length,
+        0,
+      );
+
      for (const messageList of messageLists) {
        if (messageList.messageExternalIds.length === 0) {
          continue;
@@ -111,6 +129,7 @@ export class MessagingMessageListFetchService {
          await this.cacheStorage.setAdd(
            `messages-to-import:${workspaceId}:${messageChannel.id}`,
            messageExternalIdsToImport,
+            ONE_WEEK_IN_MILLISECONDS,
          );
        }

@@ -121,9 +140,28 @@ export class MessagingMessageListFetchService {
        );
      }

+      if (totalMessageCount === 0) {
+        await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
+          [messageChannel.id],
+        );
+
+        return;
+      }
+
      await this.messageChannelSyncStatusService.scheduleMessagesImport([
        messageChannel.id,
      ]);
+
+      if (totalMessageCount < MAX_MESSAGE_COUNT_FOR_QUICK_IMPORT) {
+        await this.messagingMessagesImportService.processMessageBatchImport(
+          {
+            ...messageChannel,
+            syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
+          },
+          messageChannel.connectedAccount,
+          workspaceId,
+        );
+      }
    } catch (error) {
      await this.messageImportErrorHandlerService.handleDriverException(
        error,
@@ -1,12 +1,12 @@
 import { Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';

+import { isDefined } from 'twenty-shared/utils';
 import { Repository } from 'typeorm';

 import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
 import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
-import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util';
 import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
 import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity';
 import { TimelineActivityRepository } from 'src/modules/timeline/repositories/timeline-activity.repository';
@@ -28,46 +28,49 @@ export class MessageParticipantListener {
      participants: MessageParticipantWorkspaceEntity[];
    }>,
  ): Promise<void> {
-    // TODO: Refactor to insertTimelineActivitiesForObject once
-    for (const eventPayload of batchEvent.events) {
-      const messageParticipants = eventPayload.participants ?? [];
+    const messageObjectMetadata =
+      await this.objectMetadataRepository.findOneOrFail({
+        where: {
+          nameSingular: 'message',
+          workspaceId: batchEvent.workspaceId,
+        },
+      });

-      // TODO: move to a job?
-
-      const dataSourceSchema = getWorkspaceSchemaName(batchEvent.workspaceId);
-
-      const messageObjectMetadata =
-        await this.objectMetadataRepository.findOneOrFail({
-          where: {
-            nameSingular: 'message',
-            workspaceId: batchEvent.workspaceId,
-          },
-        });
+    const timelineActivityPayloads = batchEvent.events.flatMap((event) => {
+      const messageParticipants = event.participants ?? [];

      const messageParticipantsWithPersonId = messageParticipants.filter(
-        (participant) => participant.personId,
+        (participant) => isDefined(participant.personId),
      );

      if (messageParticipantsWithPersonId.length === 0) {
        return;
      }

-      await this.timelineActivityRepository.insertTimelineActivitiesForObject(
-        'person',
-        messageParticipantsWithPersonId.map((participant) => ({
-          dataSourceSchema,
-          name: 'message.linked',
-          properties: null,
-          objectName: 'message',
-          recordId: participant.personId,
-          workspaceMemberId: eventPayload.workspaceMemberId,
-          workspaceId: batchEvent.workspaceId,
-          linkedObjectMetadataId: messageObjectMetadata.id,
-          linkedRecordId: participant.messageId,
-          linkedRecordCachedName: '',
-        })),
-        batchEvent.workspaceId,
-      );
-    }
+      return messageParticipantsWithPersonId
+        .map((participant) => {
+          if (!isDefined(participant.personId)) {
+            return;
+          }
+
+          return {
+            name: 'message.linked',
+            properties: {},
+            objectSingularName: 'person',
+            recordId: participant.personId,
+            workspaceMemberId: event.workspaceMemberId,
+            linkedObjectMetadataId: messageObjectMetadata.id,
+            linkedRecordId: participant.messageId,
+            linkedRecordCachedName: '',
+          };
+        })
+        .filter(isDefined);
+    });
+
+    await this.timelineActivityRepository.upsertTimelineActivities({
+      objectSingularName: 'person',
+      workspaceId: batchEvent.workspaceId,
+      payloads: timelineActivityPayloads.filter(isDefined),
+    });
  }
 }
@@ -1,3 +1,6 @@
+import { isDefined } from 'twenty-shared/utils';
+import { In } from 'typeorm';
+
 import { ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
@@ -18,49 +21,62 @@ export class UpsertTimelineActivityFromInternalEvent {
  async handle(
    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordNonDestructiveEvent>,
  ): Promise<void> {
-    for (const eventData of workspaceEventBatch.events) {
-      if (eventData.userId) {
-        const workspaceMemberRepository =
-          await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-            workspaceEventBatch.workspaceId,
-            WorkspaceMemberWorkspaceEntity,
-            {
-              shouldBypassPermissionChecks: true,
-            },
-          );
-        const workspaceMember = await workspaceMemberRepository.findOneByOrFail(
-          {
-            userId: eventData.userId,
-          },
-        );
+    const workspaceMemberRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
+        workspaceEventBatch.workspaceId,
+        WorkspaceMemberWorkspaceEntity,
+        {
+          shouldBypassPermissionChecks: true,
+        },
+      );

+    const userIds = workspaceEventBatch.events
+      .map((event) => event.userId)
+      .filter(isDefined);
+
+    const workspaceMembers = await workspaceMemberRepository.findBy({
+      userId: In(userIds),
+    });
+
+    for (const eventData of workspaceEventBatch.events) {
+      const workspaceMember = workspaceMembers.find(
+        (workspaceMember) => workspaceMember.userId === eventData.userId,
+      );
+
+      if (eventData.userId && workspaceMember) {
        eventData.workspaceMemberId = workspaceMember.id;
      }
-
-      // Temporary
-      // We ignore every that is not a LinkedObject or a Business Object
-      if (
-        eventData.objectMetadata.isSystem &&
-        eventData.objectMetadata.nameSingular !== 'noteTarget' &&
-        eventData.objectMetadata.nameSingular !== 'taskTarget'
-      ) {
-        continue;
-      }
-
-      await this.timelineActivityService.upsertEvent({
-        event:
-          // we remove "before" and "after" property for a cleaner/slimmer event payload
-          'diff' in eventData.properties && eventData.properties.diff
-            ? {
-                ...eventData,
-                properties: {
-                  diff: eventData.properties.diff,
-                },
-              }
-            : eventData,
-        eventName: workspaceEventBatch.name,
-        workspaceId: workspaceEventBatch.workspaceId,
-      });
    }
+
+    const filteredEvents = workspaceEventBatch.events
+      .filter((event) => {
+        return (
+          !event.objectMetadata.isSystem ||
+          event.objectMetadata.nameSingular === 'noteTarget' ||
+          event.objectMetadata.nameSingular === 'taskTarget'
+        );
+      })
+      .map((event) => {
+        if ('diff' in event.properties && event.properties.diff) {
+          return {
+            ...event,
+            properties: {
+              diff: event.properties.diff,
+            },
+          };
+        }
+
+        return event;
+      });
+
+    if (filteredEvents.length === 0) {
+      return;
+    }
+
+    await this.timelineActivityService.upsertEvents({
+      events: filteredEvents,
+      eventName: workspaceEventBatch.name,
+      workspaceId: workspaceEventBatch.workspaceId,
+    });
  }
 }
@@ -1,11 +1,19 @@
 import { Injectable } from '@nestjs/common';

-import { MoreThan } from 'typeorm';
+import { isDefined } from 'class-validator';
+import { In, MoreThan } from 'typeorm';

 import { ObjectRecord } from 'src/engine/api/graphql/workspace-query-builder/interfaces/object-record.interface';

 import { objectRecordDiffMerge } from 'src/engine/core-modules/event-emitter/utils/object-record-diff-merge';
 import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import { TimelineActivityPayload } from 'src/modules/timeline/types/timeline-activity-payload';
+
+type TimelineActivityPayloadWorkspaceIdAndObjectSingularName = {
+  payloads: TimelineActivityPayload[];
+  workspaceId: string;
+  objectSingularName: string;
+};

@Injectable()
 export class TimelineActivityRepository {
@@ -13,82 +21,67 @@ export class TimelineActivityRepository {
    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
  ) {}

-  async upsertOne({
-    name,
-    objectName,
-    properties,
-    recordId,
+  async upsertTimelineActivities({
+    objectSingularName,
    workspaceId,
-    linkedObjectMetadataId,
-    linkedRecordCachedName,
-    linkedRecordId,
-    workspaceMemberId,
-  }: {
-    name: string;
-    properties: Partial<ObjectRecord>;
-    objectName: string;
-    recordId: string;
-    workspaceId: string;
-    workspaceMemberId?: string;
-    linkedRecordCachedName?: string;
-    linkedRecordId?: string;
-    linkedObjectMetadataId: string | null;
-  }) {
-    const recentTimelineActivity = await this.findRecentTimelineActivity(
-      name,
-      objectName,
-      recordId,
-      workspaceMemberId,
-      linkedRecordId,
+    payloads,
+  }: TimelineActivityPayloadWorkspaceIdAndObjectSingularName) {
+    const recentTimelineActivities = await this.findRecentTimelineActivities({
+      objectSingularName,
      workspaceId,
-    );
+      payloads,
+    });

-    // If the diff is empty, we don't need to insert or update an activity
-    // this should be handled differently, events should not be triggered when we will use proper DB events.
-    const isDiffEmpty =
-      properties.diff !== null &&
-      properties.diff &&
-      Object.keys(properties.diff).length === 0;
+    const payloadsWithDiff = payloads.filter(({ properties }) => {
+      const isDiffEmpty =
+        properties.diff !== null &&
+        properties.diff &&
+        Object.keys(properties.diff).length === 0;

-    if (isDiffEmpty) {
-      return;
-    }
+      return !isDiffEmpty;
+    });

-    if (recentTimelineActivity.length !== 0) {
-      const newProps = objectRecordDiffMerge(
-        recentTimelineActivity[0].properties,
-        properties,
+    const payloadsToInsert: TimelineActivityPayload[] = [];
+
+    for (const payload of payloadsWithDiff) {
+      const recentTimelineActivity = recentTimelineActivities.find(
+        (timelineActivity) =>
+          timelineActivity[`${objectSingularName}Id`] === payload.recordId &&
+          timelineActivity.workspaceMemberId === payload.workspaceMemberId &&
+          (!isDefined(payload.linkedRecordId) ||
+            timelineActivity.linkedRecordId === payload.linkedRecordId) &&
+          timelineActivity.name === payload.name,
      );

-      return this.updateTimelineActivity(
-        recentTimelineActivity[0].id,
-        newProps,
-        workspaceMemberId,
-        workspaceId,
-      );
+      if (recentTimelineActivity) {
+        const mergedProperties = objectRecordDiffMerge(
+          recentTimelineActivity.properties,
+          payload.properties,
+        );
+
+        await this.updateTimelineActivity({
+          id: recentTimelineActivity.id,
+          properties: mergedProperties,
+          workspaceMemberId: payload.workspaceMemberId,
+          workspaceId,
+        });
+      } else {
+        payloadsToInsert.push(payload);
+      }
    }

-    return this.insertTimelineActivity({
-      name,
-      properties,
-      objectName,
-      recordId,
-      workspaceMemberId,
-      linkedRecordCachedName: linkedRecordCachedName ?? '',
-      linkedRecordId,
-      linkedObjectMetadataId,
+    await this.insertTimelineActivities({
+      objectSingularName,
+      payloads: payloadsToInsert,
      workspaceId,
    });
  }

-  private async findRecentTimelineActivity(
-    name: string,
-    objectName: string,
-    recordId: string,
-    workspaceMemberId: string | undefined,
-    linkedRecordId: string | undefined,
-    workspaceId: string,
-  ) {
+  private async findRecentTimelineActivities({
+    objectSingularName,
+    workspaceId,
+    payloads,
+  }: TimelineActivityPayloadWorkspaceIdAndObjectSingularName) {
    const timelineActivityTypeORMRepository =
      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
        workspaceId,
@@ -101,31 +94,65 @@ export class TimelineActivityRepository {
    const tenMinutesAgo = new Date(Date.now() - 10 * 60 * 1000);

    const whereConditions: Record<string, unknown> = {
-      [objectName + 'Id']: recordId,
-      name: name,
-      workspaceMemberId: workspaceMemberId,
+      [`${objectSingularName}Id`]: In(
+        payloads.map((payload) => payload.recordId),
+      ),
+      name: In(payloads.map((payload) => payload.name)),
+      workspaceMemberId: In(
+        payloads.map((payload) => payload.workspaceMemberId),
+      ),
      createdAt: MoreThan(tenMinutesAgo),
    };

-    if (linkedRecordId) {
-      whereConditions.linkedRecordId = linkedRecordId;
-    } else {
-      whereConditions.linkedRecordId = null;
-    }
-
-    return timelineActivityTypeORMRepository.find({
+    return await timelineActivityTypeORMRepository.find({
      where: whereConditions,
      order: { createdAt: 'DESC' },
      take: 1,
    });
  }

-  private async updateTimelineActivity(
-    id: string,
-    properties: Partial<ObjectRecord>,
-    workspaceMemberId: string | undefined,
-    workspaceId: string,
-  ) {
+  public async insertTimelineActivities({
+    objectSingularName,
+    workspaceId,
+    payloads,
+  }: TimelineActivityPayloadWorkspaceIdAndObjectSingularName) {
+    if (payloads.length === 0) {
+      return;
+    }
+
+    const timelineActivityTypeORMRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
+        workspaceId,
+        'timelineActivity',
+        {
+          shouldBypassPermissionChecks: true,
+        },
+      );
+
+    return timelineActivityTypeORMRepository.insert(
+      payloads.map((payload) => ({
+        name: payload.name,
+        properties: payload.properties,
+        workspaceMemberId: payload.workspaceMemberId,
+        [`${objectSingularName}Id`]: payload.recordId,
+        linkedRecordCachedName: payload.linkedRecordCachedName ?? '',
+        linkedRecordId: payload.linkedRecordId,
+        linkedObjectMetadataId: payload.linkedObjectMetadataId,
+      })),
+    );
+  }
+
+  private async updateTimelineActivity({
+    id,
+    properties,
+    workspaceMemberId,
+    workspaceId,
+  }: {
+    id: string;
+    properties: Partial<ObjectRecord>;
+    workspaceMemberId: string | undefined;
+    workspaceId: string;
+  }) {
    const timelineActivityTypeORMRepository =
      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
        workspaceId,
@@ -140,83 +167,4 @@ export class TimelineActivityRepository {
      workspaceMemberId: workspaceMemberId,
    });
  }
-
-  private async insertTimelineActivity({
-    linkedObjectMetadataId,
-    linkedRecordCachedName,
-    linkedRecordId,
-    name,
-    objectName,
-    properties,
-    recordId,
-    workspaceId,
-    workspaceMemberId,
-  }: {
-    name: string;
-    properties: Partial<ObjectRecord>;
-    objectName: string;
-    recordId: string;
-    workspaceMemberId: string | undefined;
-    linkedRecordCachedName: string;
-    linkedRecordId: string | undefined;
-    linkedObjectMetadataId: string | null;
-    workspaceId: string;
-  }) {
-    const timelineActivityTypeORMRepository =
-      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-        workspaceId,
-        'timelineActivity',
-        {
-          shouldBypassPermissionChecks: true,
-        },
-      );
-
-    return timelineActivityTypeORMRepository.insert({
-      name: name,
-      properties: properties,
-      workspaceMemberId: workspaceMemberId,
-      [objectName + 'Id']: recordId,
-      linkedRecordCachedName: linkedRecordCachedName ?? '',
-      linkedRecordId: linkedRecordId,
-      linkedObjectMetadataId: linkedObjectMetadataId,
-    });
-  }
-
-  public async insertTimelineActivitiesForObject(
-    objectName: string,
-    activities: {
-      name: string;
-      properties: Partial<ObjectRecord> | null;
-      workspaceMemberId: string | undefined;
-      recordId: string | null;
-      linkedRecordCachedName: string;
-      linkedRecordId: string | null | undefined;
-      linkedObjectMetadataId: string | undefined;
-    }[],
-    workspaceId: string,
-  ) {
-    if (activities.length === 0) {
-      return;
-    }
-    const timelineActivityTypeORMRepository =
-      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-        workspaceId,
-        'timelineActivity',
-        {
-          shouldBypassPermissionChecks: true,
-        },
-      );
-
-    return timelineActivityTypeORMRepository.insert(
-      activities.map((activity) => ({
-        name: activity.name,
-        properties: activity.properties,
-        workspaceMemberId: activity.workspaceMemberId,
-        [objectName + 'Id']: activity.recordId,
-        linkedRecordCachedName: activity.linkedRecordCachedName ?? '',
-        linkedRecordId: activity.linkedRecordId,
-        linkedObjectMetadataId: activity.linkedObjectMetadataId,
-      })),
-    );
-  }
 }
@@ -1,22 +1,24 @@
 import { Injectable } from '@nestjs/common';

 import { isDefined } from 'twenty-shared/utils';
+import { In } from 'typeorm';
+
+import { ObjectRecord } from 'src/engine/api/graphql/workspace-query-builder/interfaces/object-record.interface';

-import { ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
 import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
 import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import { parseEventNameOrThrow } from 'src/engine/workspace-event-emitter/utils/parse-event-name';
 import { TimelineActivityRepository } from 'src/modules/timeline/repositories/timeline-activity.repository';
 import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity';
+import { TimelineActivityPayload } from 'src/modules/timeline/types/timeline-activity-payload';

-type TimelineActivity = Omit<ObjectRecordNonDestructiveEvent, 'properties'> & {
-  name: string;
-  objectName?: string;
-  linkedRecordCachedName?: string;
-  linkedRecordId?: string;
-  linkedObjectMetadataId?: string | null;
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  properties: Record<string, any>; // more relaxed conditions than for internal events
+type ActivityType = 'note' | 'task';
+
+type EventsWithNameAndWorkspaceId = {
+  events: ObjectRecordBaseEvent[];
+  eventName: string;
+  workspaceId: string;
 };

@Injectable()
@@ -27,141 +29,138 @@ export class TimelineActivityService {
    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
  ) {}

-  private targetObjects: Record<string, string> = {
+  private targetObjects: Record<ActivityType, string> = {
    note: 'noteTarget',
    task: 'taskTarget',
  };

-  async upsertEvent({
-    event,
+  async upsertEvents({
+    events,
    eventName,
    workspaceId,
-  }: {
-    event: ObjectRecordBaseEvent;
-    eventName: string;
-    workspaceId: string;
-  }) {
-    const timelineActivities = await this.transformEventToTimelineActivities({
-      event,
-      workspaceId,
-      eventName,
-    });
+  }: EventsWithNameAndWorkspaceId) {
+    const { objectSingularName } = parseEventNameOrThrow(eventName);

-    if (!timelineActivities || timelineActivities.length === 0) return;
-
-    for (const timelineActivity of timelineActivities) {
-      const {
-        name,
-        properties,
-        recordId,
-        linkedObjectMetadataId,
-        linkedRecordCachedName,
-        linkedRecordId,
-        objectName,
-        workspaceMemberId,
-      } = timelineActivity;
-
-      await this.timelineActivityRepository.upsertOne({
-        linkedObjectMetadataId: linkedObjectMetadataId ?? null,
-        name,
-        objectName: objectName ?? event.objectMetadata.nameSingular,
-        properties,
-        recordId,
-        workspaceId,
-        linkedRecordCachedName,
-        linkedRecordId,
-        workspaceMemberId,
-      });
-    }
-  }
-
-  private async transformEventToTimelineActivities({
-    event,
-    workspaceId,
-    eventName,
-  }: {
-    event: ObjectRecordBaseEvent;
-    workspaceId: string;
-    eventName: string;
-  }): Promise<TimelineActivity[] | undefined> {
-    if (['note', 'task'].includes(event.objectMetadata.nameSingular)) {
-      const linkedTimelineActivities = await this.getLinkedTimelineActivities({
-        event,
+    const timelineActivitiesPayloads =
+      await this.transformEventsToTimelineActivityPayloads({
+        events,
        workspaceId,
        eventName,
      });

-      // 2 timelines, one for the linked object and one for the task/note
-      if (linkedTimelineActivities && linkedTimelineActivities?.length > 0)
-        return [
-          ...linkedTimelineActivities,
-          { ...event, name: eventName },
-        ] satisfies TimelineActivity[];
+    if (
+      !timelineActivitiesPayloads ||
+      timelineActivitiesPayloads.length === 0
+    ) {
+      return;
+    }
+
+    const payloadsByObjectSingularName = timelineActivitiesPayloads.reduce(
+      (acc, payload) => {
+        const computedObjectSingularName =
+          payload.overrideObjectSingularName ?? objectSingularName;
+
+        acc[computedObjectSingularName] = [
+          ...(acc[computedObjectSingularName] || []),
+          payload,
+        ];
+
+        return acc;
+      },
+      {} as Record<string, TimelineActivityPayload[]>,
+    );
+
+    for (const objectSingularName in payloadsByObjectSingularName) {
+      this.timelineActivityRepository.upsertTimelineActivities({
+        objectSingularName,
+        workspaceId,
+        payloads: payloadsByObjectSingularName[objectSingularName],
+      });
+    }
+  }
+
+  private async transformEventsToTimelineActivityPayloads({
+    events,
+    workspaceId,
+    eventName,
+  }: EventsWithNameAndWorkspaceId): Promise<
+    TimelineActivityPayload[] | undefined
+  > {
+    const { objectSingularName } = parseEventNameOrThrow(eventName);
+
+    if (objectSingularName === 'note') {
+      const noteEventsTimelineActivities =
+        await this.computeTimelineActivityPayloadsForActivities({
+          events,
+          activityType: 'note',
+          workspaceId,
+          eventName,
+        });
+
+      return [
+        ...noteEventsTimelineActivities,
+        ...(events.map((event) => ({
+          name: eventName,
+          objectSingularName,
+          recordId: event.recordId,
+          workspaceMemberId: event.workspaceMemberId,
+          properties: event.properties,
+        })) satisfies TimelineActivityPayload[]),
+      ];
+    }
+
+    if (objectSingularName === 'task') {
+      const taskEventsTimelineActivities =
+        await this.computeTimelineActivityPayloadsForActivities({
+          events,
+          activityType: 'task',
+          workspaceId,
+          eventName,
+        });
+
+      return [
+        ...taskEventsTimelineActivities,
+        ...(events.map((event) => ({
+          name: eventName,
+          objectSingularName,
+          recordId: event.recordId,
+          workspaceMemberId: event.workspaceMemberId,
+          properties: event.properties,
+        })) satisfies TimelineActivityPayload[]),
+      ];
    }

    if (
-      ['noteTarget', 'taskTarget', 'messageParticipant'].includes(
-        event.objectMetadata.nameSingular,
-      )
+      objectSingularName === 'noteTarget' ||
+      objectSingularName === 'taskTarget'
    ) {
-      return await this.getLinkedTimelineActivities({
-        event,
+      return await this.computeTimelineActivityPayloadsForActivityTargets({
+        events,
+        activityType: objectSingularName === 'noteTarget' ? 'note' : 'task',
        workspaceId,
        eventName,
      });
    }

-    return [{ ...event, name: eventName }] satisfies TimelineActivity[];
+    return events.map((event) => ({
+      name: eventName,
+      objectSingularName,
+      recordId: event.recordId,
+      workspaceMemberId: event.workspaceMemberId,
+      properties: event.properties,
+    })) satisfies TimelineActivityPayload[];
  }

-  private async getLinkedTimelineActivities({
-    event,
-    workspaceId,
-    eventName,
-  }: {
-    event: ObjectRecordBaseEvent;
-    workspaceId: string;
-    eventName: string;
-  }): Promise<TimelineActivity[] | undefined> {
-    switch (event.objectMetadata.nameSingular) {
-      case 'noteTarget':
-        return this.computeActivityTargets({
-          event,
-          activityType: 'note',
-          eventName,
-          workspaceId,
-        });
-      case 'taskTarget':
-        return this.computeActivityTargets({
-          event,
-          activityType: 'task',
-          eventName,
-          workspaceId,
-        });
-      case 'note':
-      case 'task':
-        return this.computeActivities({
-          event,
-          activityType: event.objectMetadata.nameSingular,
-          eventName,
-          workspaceId,
-        });
-      default:
-        return [];
-    }
-  }
-
-  private async computeActivities({
-    event,
+  private async computeTimelineActivityPayloadsForActivities({
+    events,
    activityType,
    eventName,
    workspaceId,
-  }: {
-    event: ObjectRecordBaseEvent;
-    activityType: string;
-    eventName: string;
-    workspaceId: string;
-  }) {
+  }: EventsWithNameAndWorkspaceId & { activityType: ActivityType }): Promise<
+    TimelineActivityPayload[]
+  > {
+    const { action } = parseEventNameOrThrow(eventName);
+
    const activityTargetRepository =
      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
        workspaceId,
@@ -173,84 +172,70 @@ export class TimelineActivityService {

    const activityTargets = await activityTargetRepository.find({
      where: {
-        [activityType + 'Id']: event.recordId,
+        [`${activityType}Id`]: In(events.map((event) => event.recordId)),
      },
    });

-    const activityRepository =
-      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-        workspaceId,
-        activityType,
-        {
-          shouldBypassPermissionChecks: true,
-        },
-      );
+    if (activityTargets.length === 0) {
+      return [];
+    }

-    const activity = await activityRepository.findOneBy({
-      id: event.recordId,
-    });
+    return events
+      .flatMap((event) => {
+        const correspondingActivityTargets = activityTargets.filter(
+          (activityTarget) =>
+            activityTarget[`${activityType}Id`] === event.recordId,
+        );

-    if (activityTargets.length === 0) return;
-    if (!isDefined(activity)) return;
+        if (correspondingActivityTargets.length === 0) {
+          return;
+        }

-    return activityTargets
-      .map((activityTarget) => {
-        const targetColumn: string[] = Object.entries(activityTarget)
-          .map(([columnName, columnValue]: [string, string]) => {
-            if (
-              columnName === activityType + 'Id' ||
-              !columnName.endsWith('Id')
-            )
-              return;
-            if (columnValue === null) return;
+        return correspondingActivityTargets.map((activityTarget) => {
+          const targetColumn: string | undefined = Object.entries(
+            activityTarget,
+          ).find(
+            ([columnName, columnValue]: [string, string]) =>
+              columnName !== activityType + 'Id' &&
+              columnName.endsWith('Id') &&
+              columnValue !== null,
+          )?.[0];

-            return columnName;
-          })
-          .filter((column): column is string => column !== undefined);
+          if (!isDefined(targetColumn)) {
+            return;
+          }

-        if (targetColumn.length === 0) return;
+          const activityTitle = (event.properties.after as ObjectRecord)?.title;
+          const activityId = event.recordId;

-        return {
-          ...event,
-          name: 'linked-' + eventName,
-          objectName: targetColumn[0].replace(/Id$/, ''),
-          recordId: activityTarget[targetColumn[0]],
-          linkedRecordCachedName: activity.title,
-          linkedRecordId: activity.id,
-          linkedObjectMetadataId: event.objectMetadata.id,
-        } satisfies TimelineActivity;
+          if (!isDefined(activityTitle)) {
+            return;
+          }
+
+          return {
+            name: `linked-${activityType}.${action}`,
+            workspaceMemberId: event.workspaceMemberId ?? '',
+            recordId: activityTarget[targetColumn.replace(/Id$/, '')],
+            linkedRecordCachedName: activityTitle,
+            linkedRecordId: activityId,
+            linkedObjectMetadataId: event.objectMetadata.id,
+            properties: event.properties,
+            overrideObjectSingularName: event.objectMetadata.nameSingular,
+          } satisfies TimelineActivityPayload;
+        });
      })
-      .filter(
-        // @ts-expect-error legacy noImplicitAny
-        (event): event is TimelineActivity => event !== undefined,
-      ) as TimelineActivity[];
+      .filter(isDefined);
  }

-  private async computeActivityTargets({
-    event,
+  private async computeTimelineActivityPayloadsForActivityTargets({
+    events,
    activityType,
    eventName,
    workspaceId,
-  }: {
-    event: ObjectRecordBaseEvent;
-    activityType: 'task' | 'note';
-    eventName: string;
-    workspaceId: string;
-  }): Promise<TimelineActivity[] | undefined> {
-    const activityTargetRepository =
-      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-        workspaceId,
-        this.targetObjects[activityType],
-        {
-          shouldBypassPermissionChecks: true,
-        },
-      );
-
-    const activityTarget = await activityTargetRepository.findOneBy({
-      id: event.recordId,
-    });
-
-    if (!isDefined(activityTarget)) return;
+  }: EventsWithNameAndWorkspaceId & { activityType: ActivityType }): Promise<
+    TimelineActivityPayload[]
+  > {
+    const { action } = parseEventNameOrThrow(eventName);

    const activityRepository =
      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
@@ -261,39 +246,86 @@ export class TimelineActivityService {
        },
      );

-    const activity = await activityRepository.findOneBy({
-      id: activityTarget.activityId,
+    const activities = await activityRepository.find({
+      where: {
+        id: In(
+          events
+            .map((event) =>
+              this.extractActivityIdFromActivityTargetEvent(
+                event,
+                activityType,
+              ),
+            )
+            .filter(isDefined),
+        ),
+      },
    });

-    if (!isDefined(activity)) return;
+    if (activities.length === 0) {
+      return [];
+    }

-    const activityObjectMetadataId = event.objectMetadata.fields.find(
-      (field) => field.name === activityType,
-    )?.relationTargetObjectMetadataId;
+    return events
+      .map((event) => {
+        const activity = activities.find(
+          (activity) =>
+            activity.id ===
+            this.extractActivityIdFromActivityTargetEvent(event, activityType),
+        );

-    const targetColumn: string[] = Object.entries(activityTarget)
-      .map(([columnName, columnValue]: [string, string]) => {
-        if (columnName === activityType + 'Id' || !columnName.endsWith('Id'))
+        if (!isDefined(activity)) {
          return;
-        if (columnValue === null) return;
+        }

-        return columnName;
+        const activityObjectMetadataId = event.objectMetadata.fields.find(
+          (field) => field.name === activityType,
+        )?.relationTargetObjectMetadataId;
+
+        if (!isDefined(activityObjectMetadataId)) {
+          return;
+        }
+
+        if (!isDefined(event.properties.after)) {
+          return;
+        }
+
+        const targetColumnName = Object.entries(event.properties.after).find(
+          ([columnName, columnValue]: [string, string]) =>
+            columnName !== activityType + 'Id' &&
+            columnName.endsWith('Id') &&
+            columnValue !== null,
+        )?.[0];
+
+        if (!isDefined(targetColumnName)) {
+          return;
+        }
+
+        const recordId = (event.properties.after as ObjectRecord)[
+          targetColumnName
+        ];
+
+        return {
+          name: `linked-${activityType}.${action}`,
+          overrideObjectSingularName: targetColumnName.replace(/Id$/, ''),
+          recordId,
+          linkedRecordCachedName: activity.title,
+          linkedRecordId: activity.id,
+          linkedObjectMetadataId: activityObjectMetadataId,
+          workspaceMemberId: event.workspaceMemberId,
+          properties: {},
+        } satisfies TimelineActivityPayload;
      })
-      .filter((column): column is string => column !== undefined);
+      .filter(isDefined);
+  }

-    if (targetColumn.length === 0) return;
-
-    return [
-      {
-        ...event,
-        name: 'linked-' + eventName,
-        properties: {},
-        objectName: targetColumn[0].replace(/Id$/, ''),
-        recordId: activityTarget[targetColumn[0]],
-        linkedRecordCachedName: activity.title,
-        linkedRecordId: activity.id,
-        linkedObjectMetadataId: activityObjectMetadataId,
-      } satisfies TimelineActivity,
+  private extractActivityIdFromActivityTargetEvent(
+    event: ObjectRecordBaseEvent,
+    activityType: ActivityType,
+  ): string | undefined {
+    const activityId = (event.properties.after as ObjectRecord)?.[
+      `${activityType}Id`
    ];
+
+    return activityId;
  }
 }
@@ -0,0 +1,10 @@
+export type TimelineActivityPayload = {
+  properties: Record<string, unknown>;
+  linkedObjectMetadataId?: string;
+  linkedRecordId?: string;
+  linkedRecordCachedName?: string;
+  workspaceMemberId?: string;
+  name: string;
+  recordId: string;
+  overrideObjectSingularName?: string;
+};