diff --git a/packages/twenty-front/src/modules/accounts/types/CalendarChannel.ts b/packages/twenty-front/src/modules/accounts/types/CalendarChannel.ts index 9664352e7c..0973f44b0a 100644 --- a/packages/twenty-front/src/modules/accounts/types/CalendarChannel.ts +++ b/packages/twenty-front/src/modules/accounts/types/CalendarChannel.ts @@ -9,10 +9,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', } diff --git a/packages/twenty-front/src/modules/accounts/types/MessageChannel.ts b/packages/twenty-front/src/modules/accounts/types/MessageChannel.ts index e39a58c525..c903d5b8f6 100644 --- a/packages/twenty-front/src/modules/accounts/types/MessageChannel.ts +++ b/packages/twenty-front/src/modules/accounts/types/MessageChannel.ts @@ -17,10 +17,13 @@ export enum MessageChannelSyncStatus { } export enum MessageChannelSyncStage { - FULL_MESSAGE_LIST_FETCH_PENDING = 'FULL_MESSAGE_LIST_FETCH_PENDING', - PARTIAL_MESSAGE_LIST_FETCH_PENDING = 'PARTIAL_MESSAGE_LIST_FETCH_PENDING', + 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', } diff --git a/packages/twenty-front/src/modules/favorites/components/FavoriteNavigationDrawerItem.tsx b/packages/twenty-front/src/modules/favorites/components/FavoriteNavigationDrawerItem.tsx index 5cf693bfce..df5b5305f0 100644 --- a/packages/twenty-front/src/modules/favorites/components/FavoriteNavigationDrawerItem.tsx +++ b/packages/twenty-front/src/modules/favorites/components/FavoriteNavigationDrawerItem.tsx @@ -28,7 +28,7 @@ export const FavoriteNavigationDrawerItem = ({ triggerEvent, }: FavoriteNavigationDrawerItemProps) => { const { objectMetadataItem } = useObjectMetadataItem({ - objectNameSingular: favorite.objectNameSingular || '', + objectNameSingular: favorite.objectNameSingular, }); if (!favorite.objectNameSingular) { diff --git a/packages/twenty-front/src/modules/favorites/utils/sortFavorites.ts b/packages/twenty-front/src/modules/favorites/utils/sortFavorites.ts index 2568a33db6..3ec210d87f 100644 --- a/packages/twenty-front/src/modules/favorites/utils/sortFavorites.ts +++ b/packages/twenty-front/src/modules/favorites/utils/sortFavorites.ts @@ -11,7 +11,7 @@ import { getAppPath } from '~/utils/navigation/getAppPath'; export type ProcessedFavorite = Favorite & { Icon?: string; - objectNameSingular?: string; + objectNameSingular: string; }; export const sortFavorites = ( @@ -24,7 +24,7 @@ export const sortFavorites = ( hasLinkToShowPage: boolean, views: Pick[], objectMetadataItems: ObjectMetadataItem[], -) => { +): ProcessedFavorite[] => { return favorites .map((favorite) => { if ( @@ -36,7 +36,8 @@ export const sortFavorites = ( if (!isDefined(view)) { return { ...favorite, - } as ProcessedFavorite; + objectNameSingular: 'view', + }; } const { namePlural } = getObjectMetadataNamePluralFromViewId( diff --git a/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsCalendars.stories.tsx b/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsCalendars.stories.tsx index fc184ea50d..fff653c437 100644 --- a/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsCalendars.stories.tsx +++ b/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsCalendars.stories.tsx @@ -95,7 +95,7 @@ export const TwoConnectedAccounts: Story = { connectedAccountId: '20202020-954c-4d76-9a87-e5f072d4b7ef', contactAutoCreationPolicy: 'SENT', - syncStage: 'PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING', + syncStage: 'FULL_CALENDAR_EVENT_LIST_FETCH_PENDING', type: 'email', isContactAutoCreationEnabled: true, syncCursor: '1562764', diff --git a/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsEmails.stories.tsx b/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsEmails.stories.tsx index f0ebac280c..0fcb79ccbd 100644 --- a/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsEmails.stories.tsx +++ b/packages/twenty-front/src/pages/settings/accounts/__stories__/SettingsAccountsEmails.stories.tsx @@ -1,5 +1,5 @@ import { Meta, StoryObj } from '@storybook/react'; -import { graphql, HttpResponse } from 'msw'; +import { HttpResponse, graphql } from 'msw'; import { PageDecorator, @@ -96,7 +96,7 @@ export const TwoConnectedAccounts: Story = { connectedAccountId: '20202020-954c-4d76-9a87-e5f072d4b7ef', contactAutoCreationPolicy: 'SENT', - syncStage: 'PARTIAL_MESSAGE_LIST_FETCH_PENDING', + syncStage: 'FULL_MESSAGE_LIST_FETCH_PENDING', type: 'email', isContactAutoCreationEnabled: true, syncCursor: '1562764', @@ -122,7 +122,7 @@ export const TwoConnectedAccounts: Story = { connectedAccountId: '20202020-954c-4d76-9a87-e5f072d4b7ef', contactAutoCreationPolicy: 'SENT', - syncStage: 'PARTIAL_MESSAGE_LIST_FETCH_PENDING', + syncStage: 'FULL_MESSAGE_LIST_FETCH_PENDING', type: 'email', isContactAutoCreationEnabled: true, syncCursor: '1562764', diff --git a/packages/twenty-front/src/testing/mock-data/generated/mock-metadata-query-result.ts b/packages/twenty-front/src/testing/mock-data/generated/mock-metadata-query-result.ts index a014f21db2..b0cbb2d633 100644 --- a/packages/twenty-front/src/testing/mock-data/generated/mock-metadata-query-result.ts +++ b/packages/twenty-front/src/testing/mock-data/generated/mock-metadata-query-result.ts @@ -16816,7 +16816,7 @@ export const mockedStandardObjectMetadataQueryResult: ObjectMetadataItemsQuery = "id": "3f99df3c-f124-491e-a9e5-deb0922f250f", "color": "blue", "label": "Partial calendar event list fetch pending", - "value": "PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING", + "value": "FULL_CALENDAR_EVENT_LIST_FETCH_PENDING", "position": 1 }, { diff --git a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts index 22a9042901..4bd6a21939 100644 --- a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts +++ b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts @@ -44,6 +44,7 @@ import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/typ import { WorkspaceDataSource } from 'src/engine/twenty-orm/datasource/workspace.datasource'; import { DeepPartialWithNestedRelationFields } from 'src/engine/twenty-orm/entity-manager/types/deep-partial-entity-with-nested-relation-fields.type'; import { QueryDeepPartialEntityWithNestedRelationFields } from 'src/engine/twenty-orm/entity-manager/types/query-deep-partial-entity-with-nested-relation-fields.type'; +import { computeTwentyORMException } from 'src/engine/twenty-orm/error-handling/compute-twenty-orm-exception'; import { RelationNestedQueries } from 'src/engine/twenty-orm/relation-nested-queries/relation-nested-queries'; import { OperationType, @@ -54,7 +55,6 @@ import { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace. import { formatData } from 'src/engine/twenty-orm/utils/format-data.util'; import { formatResult } from 'src/engine/twenty-orm/utils/format-result.util'; import { getObjectMetadataFromEntityTarget } from 'src/engine/twenty-orm/utils/get-object-metadata-from-entity-target.util'; -import { computeTwentyORMException } from 'src/engine/twenty-orm/error-handling/compute-twenty-orm-exception'; type PermissionOptions = { shouldBypassPermissionChecks?: boolean; @@ -173,6 +173,7 @@ export class WorkspaceEntityManager extends EntityManager { | QueryDeepPartialEntityWithNestedRelationFields[], selectedColumns: string[] | '*' = '*', permissionOptions?: PermissionOptions, + authContext?: AuthContext, ): Promise { const metadata = this.connection.getMetadata(target); @@ -183,6 +184,7 @@ export class WorkspaceEntityManager extends EntityManager { permissionOptions, ) .insert() + .setAuthContext(authContext ?? {}) .values(entity) .returning(selectedColumns) .execute(); diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/__tests__/workspace.repository.spec.ts b/packages/twenty-server/src/engine/twenty-orm/repository/__tests__/workspace.repository.spec.ts index bcd2eaa960..a081ef00ae 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/__tests__/workspace.repository.spec.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/__tests__/workspace.repository.spec.ts @@ -273,6 +273,7 @@ describe('WorkspaceRepository', () => { shouldBypassPermissionChecks: false, objectRecordsPermissions: mockObjectRecordsPermissions, }, + undefined, ); }); diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts index 12cf1e19a6..6791a04aac 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts @@ -251,4 +251,10 @@ export class WorkspaceInsertQueryBuilder< TwentyORMExceptionCode.METHOD_NOT_ALLOWED, ); } + + setAuthContext(authContext: AuthContext): WorkspaceInsertQueryBuilder { + this.authContext = authContext; + + return this; + } } diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts index dac524c004..0487a78c31 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts @@ -554,6 +554,7 @@ export class WorkspaceRepository< entity, selectedColumns, permissionOptions, + this.authContext, ); } diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/utils/compute-event-name.ts b/packages/twenty-server/src/engine/workspace-event-emitter/utils/compute-event-name.ts new file mode 100644 index 0000000000..be0149421e --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-event-emitter/utils/compute-event-name.ts @@ -0,0 +1,9 @@ +import { checkStringIsDatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/utils/check-string-is-database-event-action'; + +export const computeEventName = (objectName: string, action: string) => { + if (!checkStringIsDatabaseEventAction(action)) { + throw new Error('Invalid action'); + } + + return `${objectName}.${action}`; +}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/utils/parse-event-name.ts b/packages/twenty-server/src/engine/workspace-event-emitter/utils/parse-event-name.ts new file mode 100644 index 0000000000..064ed39450 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-event-emitter/utils/parse-event-name.ts @@ -0,0 +1,20 @@ +import { isDefined } from 'twenty-shared/utils'; + +import { checkStringIsDatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/utils/check-string-is-database-event-action'; + +export const parseEventNameOrThrow = (eventName: string) => { + const [objectSingularName, action] = eventName.split('.'); + + if (!checkStringIsDatabaseEventAction(action)) { + throw new Error('Invalid event name'); + } + + if (!isDefined(objectSingularName)) { + throw new Error('Invalid event name'); + } + + return { + objectSingularName, + action, + }; +}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts index 230e2bf785..bd7d0b2e4b 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts @@ -16,6 +16,7 @@ import { objectRecordChangedValues } from 'src/engine/core-modules/event-emitter import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps'; import { CustomEventName } from 'src/engine/workspace-event-emitter/types/custom-event-name.type'; import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-ids'; +import { computeEventName } from 'src/engine/workspace-event-emitter/utils/compute-event-name'; type ActionEventMap = { [DatabaseEventAction.CREATED]: ObjectRecordCreateEvent; @@ -147,7 +148,7 @@ export class WorkspaceEventEmitter { return; } - const eventName = `${objectMetadataNameSingular}.${action}`; + const eventName = computeEventName(objectMetadataNameSingular, action); this.eventEmitter.emit(eventName, { name: eventName, diff --git a/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts b/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts index 10e060fe81..cff8548ecb 100644 --- a/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts +++ b/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts @@ -54,7 +54,7 @@ export class BlocklistReimportCalendarEventsJob { }, }); - await this.calendarChannelSyncStatusService.resetAndScheduleFullCalendarEventListFetch( + await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch( calendarChannels.map((calendarChannel) => calendarChannel.id), workspaceId, ); diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts index 06c6ed14b8..7fbab80f55 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts @@ -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, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts index c7880459da..4de0af9e7f 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts @@ -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; diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-event-import-exception-handler.service.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-event-import-exception-handler.service.ts index 0e0bb0f1a6..12752f6cb1 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-event-import-exception-handler.service.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-event-import-exception-handler.service.ts @@ -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, ); diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-events-import.service.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-events-import.service.ts index 20062f676f..6e8cb64705 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-events-import.service.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-events-import.service.ts @@ -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) { diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-fetch-events.service.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-fetch-events.service.ts index a3d536fb89..a62910c8ad 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-fetch-events.service.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-fetch-events.service.ts @@ -79,7 +79,7 @@ export class CalendarFetchEventsService { }, ); - await this.calendarChannelSyncStatusService.schedulePartialCalendarEventListFetch( + await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch( [calendarChannel.id], ); } diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts index a35426c9f6..23abc7faf7 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts @@ -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 { - 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), + }); } } diff --git a/packages/twenty-server/src/modules/calendar/common/services/calendar-channel-sync-status.service.ts b/packages/twenty-server/src/modules/calendar/common/services/calendar-channel-sync-status.service.ts index ac2385d5df..9a754631ec 100644 --- a/packages/twenty-server/src/modules/calendar/common/services/calendar-channel-sync-status.service.ts +++ b/packages/twenty-server/src/modules/calendar/common/services/calendar-channel-sync-status.service.ts @@ -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( - '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, diff --git a/packages/twenty-server/src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity.ts b/packages/twenty-server/src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity.ts index 5abfb5df11..a827376a4d 100644 --- a/packages/twenty-server/src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity.ts +++ b/packages/twenty-server/src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity.ts @@ -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}'`, }) diff --git a/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts b/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts index 07551f05b2..c5bbc03345 100644 --- a/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts +++ b/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts @@ -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( 'messageChannel', diff --git a/packages/twenty-server/src/modules/messaging/common/standard-objects/message-channel.workspace-entity.ts b/packages/twenty-server/src/modules/messaging/common/standard-objects/message-channel.workspace-entity.ts index 23c6f8cffb..638f0f1cdb 100644 --- a/packages/twenty-server/src/modules/messaging/common/standard-objects/message-channel.workspace-entity.ts +++ b/packages/twenty-server/src/modules/messaging/common/standard-objects/message-channel.workspace-entity.ts @@ -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}'`, }) diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts index da57b075aa..155c462f85 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts @@ -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, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts index a8e2bd3e72..533580b338 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts @@ -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, workspaceId: string, ): Promise { - await this.messageChannelSyncStatusService.markAsFailedAndFlushMessagesToImport( + await this.messageChannelSyncStatusService.markAsFailed( [messageChannel.id], workspaceId, MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS, @@ -184,7 +184,7 @@ export class MessageImportExceptionHandlerService { messageChannel: Pick, workspaceId: string, ): Promise { - await this.messageChannelSyncStatusService.markAsFailedAndFlushMessagesToImport( + await this.messageChannelSyncStatusService.markAsFailed( [messageChannel.id], workspaceId, MessageChannelSyncStatus.FAILED_UNKNOWN, @@ -213,7 +213,7 @@ export class MessageImportExceptionHandlerService { messageChannel: Pick, workspaceId: string, ): Promise { - await this.messageChannelSyncStatusService.markAsFailedAndFlushMessagesToImport( + await this.messageChannelSyncStatusService.markAsFailed( [messageChannel.id], workspaceId, MessageChannelSyncStatus.FAILED_UNKNOWN, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts index c7f21371b8..3d78cc640a 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts @@ -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, ); + messagingGetMessageListService = module.get( MessagingGetMessageListService, ); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts index cc97a19ec4..68a4223411 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts @@ -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, diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts index ddea63bb8a..87b79c3eee 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts @@ -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 { - // 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), + }); } } diff --git a/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts b/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts index 5458809052..cb7a05cb38 100644 --- a/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts +++ b/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts @@ -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, ): Promise { - 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, + }); } } diff --git a/packages/twenty-server/src/modules/timeline/repositories/timeline-activity.repository.ts b/packages/twenty-server/src/modules/timeline/repositories/timeline-activity.repository.ts index 34ab6e78bf..b0a9028432 100644 --- a/packages/twenty-server/src/modules/timeline/repositories/timeline-activity.repository.ts +++ b/packages/twenty-server/src/modules/timeline/repositories/timeline-activity.repository.ts @@ -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; - 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 = { - [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, - 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; + 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; - 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 | 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, - })), - ); - } } diff --git a/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts b/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts index dd1c44ba2a..54f1780847 100644 --- a/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts +++ b/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts @@ -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 & { - name: string; - objectName?: string; - linkedRecordCachedName?: string; - linkedRecordId?: string; - linkedObjectMetadataId?: string | null; - // eslint-disable-next-line @typescript-eslint/no-explicit-any - properties: Record; // 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 = { + private targetObjects: Record = { 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 { - 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, + ); + + 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 { - 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 { - 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; } } diff --git a/packages/twenty-server/src/modules/timeline/types/timeline-activity-payload.ts b/packages/twenty-server/src/modules/timeline/types/timeline-activity-payload.ts new file mode 100644 index 0000000000..400958f630 --- /dev/null +++ b/packages/twenty-server/src/modules/timeline/types/timeline-activity-payload.ts @@ -0,0 +1,10 @@ +export type TimelineActivityPayload = { + properties: Record; + linkedObjectMetadataId?: string; + linkedRecordId?: string; + linkedRecordCachedName?: string; + workspaceMemberId?: string; + name: string; + recordId: string; + overrideObjectSingularName?: string; +};