From 04a370e043748e6021477f2f8daa7f921415a3f8 Mon Sep 17 00:00:00 2001 From: Lucas Bordeau Date: Mon, 12 Jan 2026 18:58:46 +0100 Subject: [PATCH] Implemented SSE subscription mechanism on the frontend (#17017) This PR is a follow-up of https://github.com/twentyhq/twenty/pull/16966 and implements a new mechanism to handle SSE events. It creates only one event stream per browser tab, then use mutations to tell the backend which query to listen to, without re-mounting the event stream connexion. Then each event that comes from this unique subscription is then dispatched in a new JavaScript CustomEvent, per queryId, on which specific hooks add an event listener. This PR introduces the generic tooling as well as the handling of update events on table. --- .../src/generated-metadata/graphql.ts | 1 + .../twenty-front/src/generated/graphql.ts | 88 +--------- .../app/components/AppRouterProviders.tsx | 10 +- .../RecordTableNoRecordGroupBody.tsx | 2 + .../components/RecordTableRow.tsx | 4 +- ...ecordTableVirtualizedDataChangedEffect.tsx | 86 ++++++--- ...eVirtualizedOnObjectRecordEventsEffect.tsx | 165 ++++++++++++++++++ .../components/ListenRecordUpdatesEffect.tsx | 2 +- .../sse-db-event/components/SSEProvider.tsx | 21 +++ .../components/SSEProviderEffect.tsx | 62 +++++++ .../components/SSEQuerySubscribeEffect.tsx | 128 ++++++++++++++ .../contexts/SseClientContext.ts | 0 .../AddQueryToEventStreamMutation.ts | 7 + .../RemoveQueryFromEventStreamMutation.ts | 9 + .../subscriptions/OnEventSubscription.ts | 25 +++ .../graphql/subscriptions/onDbEvent.ts | 0 .../hooks/__tests__/useOnDbEvent.test.tsx | 2 +- .../useListenToObjectRecordEventsForQuery.ts | 76 ++++++++ .../hooks/useOnDbEvent.ts | 4 +- .../hooks/useSseClient.util.ts | 0 .../states/activeQueryListenersState.ts | 9 + .../states/requiredQueryListenersState.ts | 9 + .../states/sseEventStreamIdState.ts | 6 + .../types/ObjectRecordEventsByQueryId.ts | 3 + .../dispatchObjectRecordEventsWithQueryIds.ts | 30 ++++ .../utils/getDefaultRecordFieldsToListen.ts} | 0 .../getObjectRecordEventsForQueryEventName.ts | 3 + .../components/SubscriptionProvider.tsx | 21 --- .../components/SubscriptionProviderEffect.tsx | 68 -------- .../subscriptions/onSubscriptionMatch.ts | 18 -- .../hooks/useSubscribeToRefetch.ts | 50 ------ .../states/subscriptionRegistryState.ts | 14 -- .../components/WorkflowRunCard.tsx | 2 +- .../dtos/object-record-event.dto.ts | 4 + .../workspace-event-emitter.resolver.ts | 10 +- .../compareArraysOfObjectsByProperty.test.ts | 92 ++++++++++ .../array/compareArraysOfObjectsByProperty.ts | 15 ++ packages/twenty-shared/src/utils/index.ts | 1 + 38 files changed, 747 insertions(+), 300 deletions(-) create mode 100644 packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedOnObjectRecordEventsEffect.tsx rename packages/twenty-front/src/modules/{subscription => sse-db-event}/components/ListenRecordUpdatesEffect.tsx (98%) create mode 100644 packages/twenty-front/src/modules/sse-db-event/components/SSEProvider.tsx create mode 100644 packages/twenty-front/src/modules/sse-db-event/components/SSEProviderEffect.tsx create mode 100644 packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx rename packages/twenty-front/src/modules/{subscription => sse-db-event}/contexts/SseClientContext.ts (100%) create mode 100644 packages/twenty-front/src/modules/sse-db-event/graphql/mutations/AddQueryToEventStreamMutation.ts create mode 100644 packages/twenty-front/src/modules/sse-db-event/graphql/mutations/RemoveQueryFromEventStreamMutation.ts create mode 100644 packages/twenty-front/src/modules/sse-db-event/graphql/subscriptions/OnEventSubscription.ts rename packages/twenty-front/src/modules/{subscription => sse-db-event}/graphql/subscriptions/onDbEvent.ts (100%) rename packages/twenty-front/src/modules/{subscription => sse-db-event}/hooks/__tests__/useOnDbEvent.test.tsx (99%) create mode 100644 packages/twenty-front/src/modules/sse-db-event/hooks/useListenToObjectRecordEventsForQuery.ts rename packages/twenty-front/src/modules/{subscription => sse-db-event}/hooks/useOnDbEvent.ts (89%) rename packages/twenty-front/src/modules/{subscription => sse-db-event}/hooks/useSseClient.util.ts (100%) create mode 100644 packages/twenty-front/src/modules/sse-db-event/states/activeQueryListenersState.ts create mode 100644 packages/twenty-front/src/modules/sse-db-event/states/requiredQueryListenersState.ts create mode 100644 packages/twenty-front/src/modules/sse-db-event/states/sseEventStreamIdState.ts create mode 100644 packages/twenty-front/src/modules/sse-db-event/types/ObjectRecordEventsByQueryId.ts create mode 100644 packages/twenty-front/src/modules/sse-db-event/utils/dispatchObjectRecordEventsWithQueryIds.ts rename packages/twenty-front/src/modules/{subscription/utils/getDefaultRecordFieldsToListen.util.ts => sse-db-event/utils/getDefaultRecordFieldsToListen.ts} (100%) create mode 100644 packages/twenty-front/src/modules/sse-db-event/utils/getObjectRecordEventsForQueryEventName.ts delete mode 100644 packages/twenty-front/src/modules/subscription/components/SubscriptionProvider.tsx delete mode 100644 packages/twenty-front/src/modules/subscription/components/SubscriptionProviderEffect.tsx delete mode 100644 packages/twenty-front/src/modules/subscription/graphql/subscriptions/onSubscriptionMatch.ts delete mode 100644 packages/twenty-front/src/modules/subscription/hooks/useSubscribeToRefetch.ts delete mode 100644 packages/twenty-front/src/modules/subscription/states/subscriptionRegistryState.ts create mode 100644 packages/twenty-shared/src/utils/array/__tests__/compareArraysOfObjectsByProperty.test.ts create mode 100644 packages/twenty-shared/src/utils/array/compareArraysOfObjectsByProperty.ts diff --git a/packages/twenty-front/src/generated-metadata/graphql.ts b/packages/twenty-front/src/generated-metadata/graphql.ts index 27be58ba62d..9ef32e8fdae 100644 --- a/packages/twenty-front/src/generated-metadata/graphql.ts +++ b/packages/twenty-front/src/generated-metadata/graphql.ts @@ -3100,6 +3100,7 @@ export type ObjectPermissionInput = { export type ObjectRecordEvent = { __typename?: 'ObjectRecordEvent'; + action: DatabaseEventAction; objectNameSingular: Scalars['String']; properties: ObjectRecordEventProperties; recordId: Scalars['String']; diff --git a/packages/twenty-front/src/generated/graphql.ts b/packages/twenty-front/src/generated/graphql.ts index 1658a874b93..9c4317463ee 100644 --- a/packages/twenty-front/src/generated/graphql.ts +++ b/packages/twenty-front/src/generated/graphql.ts @@ -3034,6 +3034,7 @@ export type ObjectPermissionInput = { export type ObjectRecordEvent = { __typename?: 'ObjectRecordEvent'; + action: DatabaseEventAction; objectNameSingular: Scalars['String']; properties: ObjectRecordEventProperties; recordId: Scalars['String']; @@ -5186,20 +5187,6 @@ export type UpdatePageLayoutWithTabsAndWidgetsMutationVariables = Exact<{ export type UpdatePageLayoutWithTabsAndWidgetsMutation = { __typename?: 'Mutation', updatePageLayoutWithTabsAndWidgets: { __typename?: 'PageLayout', id: any, name: string, type: PageLayoutType, objectMetadataId?: any | null, createdAt: string, updatedAt: string, deletedAt?: string | null, tabs?: Array<{ __typename?: 'PageLayoutTab', id: any, title: string, position: number, pageLayoutId: any, createdAt: string, updatedAt: string, widgets?: Array<{ __typename?: 'PageLayoutWidget', id: any, title: string, type: WidgetType, objectMetadataId?: any | null, createdAt: string, updatedAt: string, deletedAt?: string | null, pageLayoutTabId: any, gridPosition: { __typename?: 'GridPosition', column: number, columnSpan: number, row: number, rowSpan: number }, configuration: { __typename?: 'AggregateChartConfiguration', configurationType: WidgetConfigurationType, aggregateFieldMetadataId: any, aggregateOperation: AggregateOperations, label?: string | null, displayDataLabel?: boolean | null, format?: string | null, description?: string | null, filter?: any | null, prefix?: string | null, suffix?: string | null, timezone?: string | null, firstDayOfTheWeek?: number | null, ratioAggregateConfig?: { __typename?: 'RatioAggregateConfig', fieldMetadataId: any, optionValue: string } | null } | { __typename?: 'BarChartConfiguration', configurationType: WidgetConfigurationType, aggregateFieldMetadataId: any, aggregateOperation: AggregateOperations, primaryAxisGroupByFieldMetadataId: any, primaryAxisGroupBySubFieldName?: string | null, primaryAxisDateGranularity?: ObjectRecordGroupByDateGranularity | null, primaryAxisOrderBy?: GraphOrderBy | null, primaryAxisManualSortOrder?: Array | null, secondaryAxisGroupByFieldMetadataId?: any | null, secondaryAxisGroupBySubFieldName?: string | null, secondaryAxisGroupByDateGranularity?: ObjectRecordGroupByDateGranularity | null, secondaryAxisOrderBy?: GraphOrderBy | null, secondaryAxisManualSortOrder?: Array | null, omitNullValues?: boolean | null, axisNameDisplay?: AxisNameDisplay | null, displayDataLabel?: boolean | null, displayLegend?: boolean | null, rangeMin?: number | null, rangeMax?: number | null, color?: string | null, description?: string | null, filter?: any | null, groupMode?: BarChartGroupMode | null, layout: BarChartLayout, isCumulative?: boolean | null, timezone?: string | null, firstDayOfTheWeek?: number | null } | { __typename?: 'CalendarConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'EmailsConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'FieldConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'FieldRichTextConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'FieldsConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'FilesConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'GaugeChartConfiguration', configurationType: WidgetConfigurationType, aggregateFieldMetadataId: any, aggregateOperation: AggregateOperations, displayDataLabel?: boolean | null, color?: string | null, description?: string | null, filter?: any | null, timezone?: string | null, firstDayOfTheWeek?: number | null } | { __typename?: 'IframeConfiguration', configurationType: WidgetConfigurationType, url?: string | null } | { __typename?: 'LineChartConfiguration', configurationType: WidgetConfigurationType, aggregateFieldMetadataId: any, aggregateOperation: AggregateOperations, primaryAxisGroupByFieldMetadataId: any, primaryAxisGroupBySubFieldName?: string | null, primaryAxisDateGranularity?: ObjectRecordGroupByDateGranularity | null, primaryAxisOrderBy?: GraphOrderBy | null, primaryAxisManualSortOrder?: Array | null, secondaryAxisGroupByFieldMetadataId?: any | null, secondaryAxisGroupBySubFieldName?: string | null, secondaryAxisGroupByDateGranularity?: ObjectRecordGroupByDateGranularity | null, secondaryAxisOrderBy?: GraphOrderBy | null, secondaryAxisManualSortOrder?: Array | null, omitNullValues?: boolean | null, axisNameDisplay?: AxisNameDisplay | null, displayDataLabel?: boolean | null, displayLegend?: boolean | null, rangeMin?: number | null, rangeMax?: number | null, color?: string | null, description?: string | null, filter?: any | null, isStacked?: boolean | null, isCumulative?: boolean | null, timezone?: string | null, firstDayOfTheWeek?: number | null } | { __typename?: 'NotesConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'PieChartConfiguration', configurationType: WidgetConfigurationType, groupByFieldMetadataId: any, aggregateFieldMetadataId: any, aggregateOperation: AggregateOperations, groupBySubFieldName?: string | null, dateGranularity?: ObjectRecordGroupByDateGranularity | null, orderBy?: GraphOrderBy | null, manualSortOrder?: Array | null, displayDataLabel?: boolean | null, showCenterMetric?: boolean | null, displayLegend?: boolean | null, color?: string | null, description?: string | null, filter?: any | null, timezone?: string | null, firstDayOfTheWeek?: number | null } | { __typename?: 'StandaloneRichTextConfiguration', configurationType: WidgetConfigurationType, body: { __typename?: 'RichTextV2Body', blocknote?: string | null, markdown?: string | null } } | { __typename?: 'TasksConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'TimelineConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'ViewConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'WorkflowConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'WorkflowRunConfiguration', configurationType: WidgetConfigurationType } | { __typename?: 'WorkflowVersionConfiguration', configurationType: WidgetConfigurationType } }> | null }> | null } }; -export type OnDbEventSubscriptionVariables = Exact<{ - input: OnDbEventInput; -}>; - - -export type OnDbEventSubscription = { __typename?: 'Subscription', onDbEvent: { __typename?: 'OnDbEvent', eventDate: string, action: DatabaseEventAction, objectNameSingular: string, updatedFields?: Array | null, record: any } }; - -export type OnSubscriptionMatchSubscriptionVariables = Exact<{ - subscriptions: Array | SubscriptionInput; -}>; - - -export type OnSubscriptionMatchSubscription = { __typename?: 'Subscription', onSubscriptionMatch?: { __typename?: 'SubscriptionMatches', matches: Array<{ __typename?: 'SubscriptionMatch', subscriptionIds: Array, event: { __typename?: 'OnDbEvent', action: DatabaseEventAction, objectNameSingular: string, eventDate: string, record: any, updatedFields?: Array | null } }> } | null }; - export type ViewFieldFragmentFragment = { __typename?: 'CoreViewField', id: any, fieldMetadataId: any, viewId: any, isVisible: boolean, position: number, size: number, aggregateOperation?: AggregateOperations | null, createdAt: string, updatedAt: string, deletedAt?: string | null }; export type ViewFilterFragmentFragment = { __typename?: 'CoreViewFilter', id: any, fieldMetadataId: any, operand: ViewFilterOperand, value: any, viewFilterGroupId?: any | null, positionInViewFilterGroup?: number | null, subFieldName?: string | null, viewId: any, createdAt: string, updatedAt: string, deletedAt?: string | null }; @@ -5871,79 +5858,6 @@ export function useUpdatePageLayoutWithTabsAndWidgetsMutation(baseOptions?: Apol export type UpdatePageLayoutWithTabsAndWidgetsMutationHookResult = ReturnType; export type UpdatePageLayoutWithTabsAndWidgetsMutationResult = Apollo.MutationResult; export type UpdatePageLayoutWithTabsAndWidgetsMutationOptions = Apollo.BaseMutationOptions; -export const OnDbEventDocument = gql` - subscription OnDbEvent($input: OnDbEventInput!) { - onDbEvent(input: $input) { - eventDate - action - objectNameSingular - updatedFields - record - } -} - `; - -/** - * __useOnDbEventSubscription__ - * - * To run a query within a React component, call `useOnDbEventSubscription` and pass it any options that fit your needs. - * When your component renders, `useOnDbEventSubscription` returns an object from Apollo Client that contains loading, error, and data properties - * you can use to render your UI. - * - * @param baseOptions options that will be passed into the subscription, supported options are listed on: https://www.apollographql.com/docs/react/api/react-hooks/#options; - * - * @example - * const { data, loading, error } = useOnDbEventSubscription({ - * variables: { - * input: // value for 'input' - * }, - * }); - */ -export function useOnDbEventSubscription(baseOptions: Apollo.SubscriptionHookOptions) { - const options = {...defaultOptions, ...baseOptions} - return Apollo.useSubscription(OnDbEventDocument, options); - } -export type OnDbEventSubscriptionHookResult = ReturnType; -export type OnDbEventSubscriptionResult = Apollo.SubscriptionResult; -export const OnSubscriptionMatchDocument = gql` - subscription OnSubscriptionMatch($subscriptions: [SubscriptionInput!]!) { - onSubscriptionMatch(subscriptions: $subscriptions) { - matches { - subscriptionIds - event { - action - objectNameSingular - eventDate - record - updatedFields - } - } - } -} - `; - -/** - * __useOnSubscriptionMatchSubscription__ - * - * To run a query within a React component, call `useOnSubscriptionMatchSubscription` and pass it any options that fit your needs. - * When your component renders, `useOnSubscriptionMatchSubscription` returns an object from Apollo Client that contains loading, error, and data properties - * you can use to render your UI. - * - * @param baseOptions options that will be passed into the subscription, supported options are listed on: https://www.apollographql.com/docs/react/api/react-hooks/#options; - * - * @example - * const { data, loading, error } = useOnSubscriptionMatchSubscription({ - * variables: { - * subscriptions: // value for 'subscriptions' - * }, - * }); - */ -export function useOnSubscriptionMatchSubscription(baseOptions: Apollo.SubscriptionHookOptions) { - const options = {...defaultOptions, ...baseOptions} - return Apollo.useSubscription(OnSubscriptionMatchDocument, options); - } -export type OnSubscriptionMatchSubscriptionHookResult = ReturnType; -export type OnSubscriptionMatchSubscriptionResult = Apollo.SubscriptionResult; export const CreateCoreViewDocument = gql` mutation CreateCoreView($input: CreateViewInput!) { createCoreView(input: $input) { diff --git a/packages/twenty-front/src/modules/app/components/AppRouterProviders.tsx b/packages/twenty-front/src/modules/app/components/AppRouterProviders.tsx index d8a3ab5c382..81a9749cb49 100644 --- a/packages/twenty-front/src/modules/app/components/AppRouterProviders.tsx +++ b/packages/twenty-front/src/modules/app/components/AppRouterProviders.tsx @@ -14,7 +14,7 @@ import { ApolloCoreProvider } from '@/object-metadata/components/ApolloCoreProvi import { ObjectMetadataItemsLoadEffect } from '@/object-metadata/components/ObjectMetadataItemsLoadEffect'; import { ObjectMetadataItemsProvider } from '@/object-metadata/components/ObjectMetadataItemsProvider'; import { PrefetchDataProvider } from '@/prefetch/components/PrefetchDataProvider'; -import { SubscriptionProvider } from '@/subscription/components/SubscriptionProvider'; +import { SSEProvider } from '@/sse-db-event/components/SSEProvider'; import { SupportChatEffect } from '@/support/components/SupportChatEffect'; import { DialogManager } from '@/ui/feedback/dialog-manager/components/DialogManager'; import { DialogComponentInstanceContext } from '@/ui/feedback/dialog-manager/contexts/DialogComponentInstanceContext'; @@ -46,8 +46,8 @@ export const AppRouterProviders = () => { - - + + @@ -73,8 +73,8 @@ export const AppRouterProviders = () => { - - + + diff --git a/packages/twenty-front/src/modules/object-record/record-table/record-table-body/components/RecordTableNoRecordGroupBody.tsx b/packages/twenty-front/src/modules/object-record/record-table/record-table-body/components/RecordTableNoRecordGroupBody.tsx index d364f17fe9e..c345d175cea 100644 --- a/packages/twenty-front/src/modules/object-record/record-table/record-table-body/components/RecordTableNoRecordGroupBody.tsx +++ b/packages/twenty-front/src/modules/object-record/record-table/record-table-body/components/RecordTableNoRecordGroupBody.tsx @@ -9,6 +9,7 @@ import { RecordTableCellPortals } from '@/object-record/record-table/record-tabl import { RecordTableAggregateFooter } from '@/object-record/record-table/record-table-footer/components/RecordTableAggregateFooter'; import { isRecordTableInitialLoadingComponentState } from '@/object-record/record-table/states/isRecordTableInitialLoadingComponentState'; import { RecordTableVirtualizedDataChangedEffect } from '@/object-record/record-table/virtualization/components/RecordTableVirtualizedDataChangedEffect'; +import { RecordTableVirtualizedOnObjectRecordEventsEffect } from '@/object-record/record-table/virtualization/components/RecordTableVirtualizedOnObjectRecordEventsEffect'; import { RecordTableVirtualizedRowTreadmillEffect } from '@/object-record/record-table/virtualization/components/RecordTableVirtualizedRowTreadmillEffect'; import { useRecoilComponentValue } from '@/ui/utilities/state/component-state/hooks/useRecoilComponentValue'; @@ -37,6 +38,7 @@ export const RecordTableNoRecordGroupBody = () => { )} + ); diff --git a/packages/twenty-front/src/modules/object-record/record-table/record-table-row/components/RecordTableRow.tsx b/packages/twenty-front/src/modules/object-record/record-table/record-table-row/components/RecordTableRow.tsx index 44a4855b248..24c5ce2b485 100644 --- a/packages/twenty-front/src/modules/object-record/record-table/record-table-row/components/RecordTableRow.tsx +++ b/packages/twenty-front/src/modules/object-record/record-table/record-table-row/components/RecordTableRow.tsx @@ -11,8 +11,8 @@ import { RecordTableRowArrowKeysEffect } from '@/object-record/record-table/reco import { RecordTableRowHotkeyEffect } from '@/object-record/record-table/record-table-row/components/RecordTableRowHotkeyEffect'; import { isRecordTableRowFocusActiveComponentState } from '@/object-record/record-table/states/isRecordTableRowFocusActiveComponentState'; import { isRecordTableRowFocusedComponentFamilyState } from '@/object-record/record-table/states/isRecordTableRowFocusedComponentFamilyState'; -import { ListenRecordUpdatesEffect } from '@/subscription/components/ListenRecordUpdatesEffect'; -import { getDefaultRecordFieldsToListen } from '@/subscription/utils/getDefaultRecordFieldsToListen.util'; +import { ListenRecordUpdatesEffect } from '@/sse-db-event/components/ListenRecordUpdatesEffect'; +import { getDefaultRecordFieldsToListen } from '@/sse-db-event/utils/getDefaultRecordFieldsToListen'; import { useRecoilComponentFamilyValue } from '@/ui/utilities/state/component-state/hooks/useRecoilComponentFamilyValue'; import { useRecoilComponentValue } from '@/ui/utilities/state/component-state/hooks/useRecoilComponentValue'; diff --git a/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedDataChangedEffect.tsx b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedDataChangedEffect.tsx index 1c306c4c9ed..1f659502b23 100644 --- a/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedDataChangedEffect.tsx +++ b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedDataChangedEffect.tsx @@ -68,46 +68,74 @@ export const RecordTableVirtualizedDataChangedEffect = () => { return; } - if (lastObjectOperation.data.type === 'update-one') { - const updateInput = lastObjectOperation.data.result.updateInput; + const updatedFieldNames = new Set(); - const updatedFieldNames = Object.keys(updateInput ?? {}) ?? []; + let thereIsAnUpdateOnAFilteredField = false; + let thereIsAnUpdateOnASortedField = false; - const updatedFieldMetadataItems = activeFieldMetadataItems.filter( - (fieldMetadataItemToFilter) => - updatedFieldNames.includes(fieldMetadataItemToFilter.name) || - (fieldMetadataItemToFilter.type === FieldMetadataType.RELATION && - updatedFieldNames.includes( - `${fieldMetadataItemToFilter.name}Id`, - )), - ); + if ( + lastObjectOperation.data.type === 'update-one' || + lastObjectOperation.data.type === 'update-many' + ) { + const updateInputs = + lastObjectOperation.data.type === 'update-one' + ? [lastObjectOperation.data.result.updateInput] + : lastObjectOperation.data.result.updateInputs; - const updatedFieldMetadataItemIds = - updatedFieldMetadataItems.map(mapById); + for (const updateInput of updateInputs) { + const fieldNamesForUpdateInput = + Object.keys(updateInput ?? {}) ?? []; - const thereIsAnUpdateOnAFilteredField = currentRecordFilters.some( - (recordFilter) => - updatedFieldMetadataItemIds.includes( - recordFilter.fieldMetadataId, - ), - ); + for (const fieldName of fieldNamesForUpdateInput) { + updatedFieldNames.add(fieldName); + } - const thereIsAnUpdateOnASortedField = currentRecordSorts.some( - (recordSort) => + const updatedFieldMetadataItems = activeFieldMetadataItems.filter( + (fieldMetadataItemToFilter) => + fieldNamesForUpdateInput.includes( + fieldMetadataItemToFilter.name, + ) || + (fieldMetadataItemToFilter.type === + FieldMetadataType.RELATION && + fieldNamesForUpdateInput.includes( + `${fieldMetadataItemToFilter.name}Id`, + )), + ); + + const updatedFieldMetadataItemIds = + updatedFieldMetadataItems.map(mapById); + + const updateOnAFilteredField = currentRecordFilters.some( + (recordFilter) => + updatedFieldMetadataItemIds.includes( + recordFilter.fieldMetadataId, + ), + ); + + const updateOnASortedField = currentRecordSorts.some((recordSort) => updatedFieldMetadataItemIds.includes(recordSort.fieldMetadataId), - ); + ); - if (updatedFieldNames.includes('position')) { - resetVirtualizationBecauseDataChanged(); - } else if ( - thereIsAnUpdateOnAFilteredField || - thereIsAnUpdateOnASortedField - ) { - resetVirtualizationBecauseDataChanged(); + if (updateOnAFilteredField) { + thereIsAnUpdateOnAFilteredField = true; + } + + if (updateOnASortedField) { + thereIsAnUpdateOnASortedField = true; + } } } else { resetVirtualizationBecauseDataChanged(); } + + if (updatedFieldNames.has('position')) { + resetVirtualizationBecauseDataChanged(); + } else if ( + thereIsAnUpdateOnAFilteredField || + thereIsAnUpdateOnASortedField + ) { + resetVirtualizationBecauseDataChanged(); + } } } }, [ diff --git a/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedOnObjectRecordEventsEffect.tsx b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedOnObjectRecordEventsEffect.tsx new file mode 100644 index 00000000000..3fa32e36e27 --- /dev/null +++ b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedOnObjectRecordEventsEffect.tsx @@ -0,0 +1,165 @@ +import { triggerUpdateRecordOptimisticEffect } from '@/apollo/optimistic-effect/utils/triggerUpdateRecordOptimisticEffect'; +import { useApolloCoreClient } from '@/object-metadata/hooks/useApolloCoreClient'; +import { useObjectMetadataItems } from '@/object-metadata/hooks/useObjectMetadataItems'; +import { useGetRecordFromCache } from '@/object-record/cache/hooks/useGetRecordFromCache'; +import { getObjectTypename } from '@/object-record/cache/utils/getObjectTypename'; +import { getRecordNodeFromRecord } from '@/object-record/cache/utils/getRecordNodeFromRecord'; +import { useObjectPermissions } from '@/object-record/hooks/useObjectPermissions'; +import { useRefetchAggregateQueries } from '@/object-record/hooks/useRefetchAggregateQueries'; +import { useRegisterObjectOperation } from '@/object-record/hooks/useRegisterObjectOperation'; +import { turnSortsIntoOrderBy } from '@/object-record/object-sort-dropdown/utils/turnSortsIntoOrderBy'; +import { useRecordsFieldVisibleGqlFields } from '@/object-record/record-field/hooks/useRecordsFieldVisibleGqlFields'; +import { currentRecordFilterGroupsComponentState } from '@/object-record/record-filter-group/states/currentRecordFilterGroupsComponentState'; +import { useFilterValueDependencies } from '@/object-record/record-filter/hooks/useFilterValueDependencies'; +import { currentRecordFiltersComponentState } from '@/object-record/record-filter/states/currentRecordFiltersComponentState'; +import { useRecordIndexContextOrThrow } from '@/object-record/record-index/contexts/RecordIndexContext'; +import { currentRecordSortsComponentState } from '@/object-record/record-sort/states/currentRecordSortsComponentState'; +import { useUpsertRecordsInStore } from '@/object-record/record-store/hooks/useUpsertRecordsInStore'; +import { useRecordTableContextOrThrow } from '@/object-record/record-table/contexts/RecordTableContext'; +import { type ObjectRecord } from '@/object-record/types/ObjectRecord'; +import { useListenToObjectRecordEventsForQuery } from '@/sse-db-event/hooks/useListenToObjectRecordEventsForQuery'; +import { useRecoilComponentValue } from '@/ui/utilities/state/component-state/hooks/useRecoilComponentValue'; + +import { + computeRecordGqlOperationFilter, + isDefined, + isNonEmptyArray, +} from 'twenty-shared/utils'; +import { useDebouncedCallback } from 'use-debounce'; +import { DatabaseEventAction } from '~/generated/graphql'; + +export const RecordTableVirtualizedOnObjectRecordEventsEffect = () => { + const { objectMetadataItem } = useRecordIndexContextOrThrow(); + const { objectNameSingular } = useRecordTableContextOrThrow(); + const { registerObjectOperation } = useRegisterObjectOperation(); + const { refetchAggregateQueries } = useRefetchAggregateQueries({ + objectMetadataNamePlural: objectMetadataItem.namePlural, + }); + const apolloCoreClient = useApolloCoreClient(); + const getRecordFromCache = useGetRecordFromCache({ + objectNameSingular, + }); + const { upsertRecordsInStore } = useUpsertRecordsInStore(); + const { objectMetadataItems } = useObjectMetadataItems(); + + const recordGqlFields = useRecordsFieldVisibleGqlFields({ + objectMetadataItem, + }); + const { objectPermissionsByObjectMetadataId } = useObjectPermissions(); + + const debouncedRefetchAggregateQueries = useDebouncedCallback( + refetchAggregateQueries, + 200, + ); + + const currentRecordFilters = useRecoilComponentValue( + currentRecordFiltersComponentState, + ); + + const currentRecordSorts = useRecoilComponentValue( + currentRecordSortsComponentState, + ); + + const currentRecordFilterGroups = useRecoilComponentValue( + currentRecordFilterGroupsComponentState, + ); + + const { filterValueDependencies } = useFilterValueDependencies(); + + const queryId = `record-table-virtualized-${objectMetadataItem.nameSingular}`; + + useListenToObjectRecordEventsForQuery({ + queryId, + operationSignature: { + objectNameSingular: objectMetadataItem.nameSingular, + variables: { + filter: computeRecordGqlOperationFilter({ + fields: objectMetadataItem.fields, + recordFilters: currentRecordFilters, + recordFilterGroups: currentRecordFilterGroups, + filterValueDependencies, + }), + orderBy: turnSortsIntoOrderBy(objectMetadataItem, currentRecordSorts), + }, + }, + onObjectRecordEvents: (objectRecordEvents) => { + const cache = apolloCoreClient.cache; + + const updateEvents = objectRecordEvents.filter( + (objectRecordEvent) => + objectRecordEvent.action === DatabaseEventAction.UPDATED, + ); + + const updatedRecordsWithUpdatedFieldsOnly: ObjectRecord[] = []; + for (const updateEvent of updateEvents) { + const updatedRecord = updateEvent.properties.after; + + upsertRecordsInStore({ partialRecords: [updatedRecord] }); + + const cachedRecord = getRecordFromCache(updatedRecord.id); + const cachedRecordWithConnection = + getRecordNodeFromRecord({ + record: cachedRecord, + objectMetadataItem, + objectMetadataItems, + recordGqlFields: recordGqlFields, + computeReferences: false, + }); + + const computedOptimisticRecord = { + ...updatedRecord, + id: updatedRecord.id, + __typename: getObjectTypename(objectMetadataItem.nameSingular), + }; + + const computedOptimisticRecordWithConnection = + getRecordNodeFromRecord({ + record: computedOptimisticRecord, + objectMetadataItem, + objectMetadataItems, + recordGqlFields: recordGqlFields, + }); + + if ( + !isDefined(cachedRecordWithConnection) || + !isDefined(computedOptimisticRecordWithConnection) + ) { + continue; + } + + triggerUpdateRecordOptimisticEffect({ + cache, + objectMetadataItem, + currentRecord: cachedRecordWithConnection, + updatedRecord: computedOptimisticRecordWithConnection, + objectMetadataItems, + objectPermissionsByObjectMetadataId, + upsertRecordsInStore, + }); + + const updatedFields = updateEvent.properties?.updatedFields ?? []; + + updatedRecordsWithUpdatedFieldsOnly.push({ + ...Object.fromEntries( + updatedFields.map((fieldName) => { + return [fieldName, updatedRecord[fieldName]]; + }), + ), + id: updatedRecord.id, + __typename: getObjectTypename(objectMetadataItem.nameSingular), + }); + } + + registerObjectOperation(objectMetadataItem, { + type: 'update-many', + result: { updateInputs: updatedRecordsWithUpdatedFieldsOnly }, + }); + + if (isNonEmptyArray(updateEvents)) { + debouncedRefetchAggregateQueries(); + } + }, + }); + + return null; +}; diff --git a/packages/twenty-front/src/modules/subscription/components/ListenRecordUpdatesEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/ListenRecordUpdatesEffect.tsx similarity index 98% rename from packages/twenty-front/src/modules/subscription/components/ListenRecordUpdatesEffect.tsx rename to packages/twenty-front/src/modules/sse-db-event/components/ListenRecordUpdatesEffect.tsx index ae7f6448463..20cdf7c555a 100644 --- a/packages/twenty-front/src/modules/subscription/components/ListenRecordUpdatesEffect.tsx +++ b/packages/twenty-front/src/modules/sse-db-event/components/ListenRecordUpdatesEffect.tsx @@ -11,7 +11,7 @@ import { useObjectPermissions } from '@/object-record/hooks/useObjectPermissions import { useUpsertRecordsInStore } from '@/object-record/record-store/hooks/useUpsertRecordsInStore'; import { recordStoreFamilyState } from '@/object-record/record-store/states/recordStoreFamilyState'; import { type ObjectRecord } from '@/object-record/types/ObjectRecord'; -import { useOnDbEvent } from '@/subscription/hooks/useOnDbEvent'; +import { useOnDbEvent } from '@/sse-db-event/hooks/useOnDbEvent'; import { useRecoilCallback } from 'recoil'; import { isDefined } from 'twenty-shared/utils'; import { DatabaseEventAction } from '~/generated/graphql'; diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEProvider.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEProvider.tsx new file mode 100644 index 00000000000..6884697a3fa --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEProvider.tsx @@ -0,0 +1,21 @@ +import { SSEProviderEffect } from '@/sse-db-event/components/SSEProviderEffect'; +import { SSEQuerySubscribeEffect } from '@/sse-db-event/components/SSEQuerySubscribeEffect'; +import { SseClientContext } from '@/sse-db-event/contexts/SseClientContext'; +import { useSseClient } from '@/sse-db-event/hooks/useSseClient.util'; +import { type ReactNode } from 'react'; + +type SSEProviderProps = { + children: ReactNode; +}; + +export const SSEProvider = ({ children }: SSEProviderProps) => { + const { sseClient } = useSseClient(); + + return ( + + + + {children} + + ); +}; diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEProviderEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEProviderEffect.tsx new file mode 100644 index 00000000000..b9806d08f19 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEProviderEffect.tsx @@ -0,0 +1,62 @@ +import { SseClientContext } from '@/sse-db-event/contexts/SseClientContext'; +import { ON_EVENT_SUBSCRIPTION } from '@/sse-db-event/graphql/subscriptions/OnEventSubscription'; +import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdState'; +import { dispatchObjectRecordEventsWithQueryIds } from '@/sse-db-event/utils/dispatchObjectRecordEventsWithQueryIds'; +import { isNonEmptyString } from '@sniptt/guards'; +import { type ExecutionResult, print } from 'graphql'; +import { useContext, useEffect } from 'react'; +import { useRecoilState } from 'recoil'; +import { isDefined } from 'twenty-shared/utils'; +import { v4 } from 'uuid'; +import { type EventSubscription } from '~/generated/graphql'; + +export const SSEProviderEffect = () => { + const sseClient = useContext(SseClientContext); + + const [sseEventStreamId, setSseEventStreamId] = useRecoilState( + sseEventStreamIdState, + ); + + useEffect(() => { + if (!isDefined(sseClient)) { + return; + } + + if (!isNonEmptyString(sseEventStreamId)) { + setSseEventStreamId(v4()); + return; + } + + const unsubscribe = sseClient.subscribe( + { + query: print(ON_EVENT_SUBSCRIPTION), + variables: { + eventStreamId: sseEventStreamId, + }, + }, + { + next: ( + value: ExecutionResult<{ onEventSubscription: EventSubscription }>, + ) => { + const objectRecordEventsWithQueryIds = + value?.data?.onEventSubscription?.eventWithQueryIdsList ?? []; + + dispatchObjectRecordEventsWithQueryIds( + objectRecordEventsWithQueryIds, + ); + }, + error: (error) => { + // eslint-disable-next-line no-console + console.error('Subscription error:', error); + }, + complete: () => {}, + }, + ); + + return () => { + unsubscribe(); + }; + }, [sseClient, sseEventStreamId, setSseEventStreamId]); + + return null; +}; diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx new file mode 100644 index 00000000000..a75cc51f27b --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx @@ -0,0 +1,128 @@ +import { useApolloCoreClient } from '@/object-metadata/hooks/useApolloCoreClient'; +import { ADD_QUERY_TO_EVENT_STREAM_MUTATION } from '@/sse-db-event/graphql/mutations/AddQueryToEventStreamMutation'; +import { REMOVE_QUERY_FROM_EVENT_STREAM_MUTATION } from '@/sse-db-event/graphql/mutations/RemoveQueryFromEventStreamMutation'; +import { activeQueryListenersState } from '@/sse-db-event/states/activeQueryListenersState'; +import { requiredQueryListenersState } from '@/sse-db-event/states/requiredQueryListenersState'; +import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdState'; +import { getSnapshotValue } from '@/ui/utilities/state/utils/getSnapshotValue'; +import { useMutation } from '@apollo/client'; +import { useEffect } from 'react'; +import { useRecoilCallback, useRecoilValue } from 'recoil'; +import { + compareArraysOfObjectsByProperty, + isDefined, +} from 'twenty-shared/utils'; +import { useDebouncedCallback } from 'use-debounce'; +import { + type AddQuerySubscriptionInput, + type RemoveQueryFromEventStreamInput, +} from '~/generated/graphql'; + +export const SSEQuerySubscribeEffect = () => { + const sseEventStreamId = useRecoilValue(sseEventStreamIdState); + + const apolloCoreClient = useApolloCoreClient(); + + const [addQueryToEventStream] = useMutation< + boolean, + { input: AddQuerySubscriptionInput } + >(ADD_QUERY_TO_EVENT_STREAM_MUTATION, { client: apolloCoreClient }); + + const [removeQueryFromEventStream] = useMutation< + void, + { input: RemoveQueryFromEventStreamInput } + >(REMOVE_QUERY_FROM_EVENT_STREAM_MUTATION, { client: apolloCoreClient }); + + const requiredQueryListeners = useRecoilValue(requiredQueryListenersState); + const activeQueryListeners = useRecoilValue(activeQueryListenersState); + + const updateQueryListeners = useRecoilCallback( + ({ set, snapshot }) => + async () => { + if (!isDefined(sseEventStreamId)) { + return; + } + + const requiredQueryListeners = getSnapshotValue( + snapshot, + requiredQueryListenersState, + ); + + const activeQueryListeners = getSnapshotValue( + snapshot, + activeQueryListenersState, + ); + + const queryListenersToAdd = requiredQueryListeners.filter( + (listener) => + !activeQueryListeners.some( + (activeListener) => activeListener.queryId === listener.queryId, + ), + ); + + const queryListenersToRemove = activeQueryListeners.filter( + (listener) => + !requiredQueryListeners.some( + (requiredListener) => + requiredListener.queryId === listener.queryId, + ), + ); + + for (const queryListenerToAdd of queryListenersToAdd) { + await addQueryToEventStream({ + variables: { + input: { + eventStreamId: sseEventStreamId, + queryId: queryListenerToAdd.queryId, + operationSignature: queryListenerToAdd.operationSignature, + }, + }, + }); + } + + for (const queryListenerToRemove of queryListenersToRemove) { + await removeQueryFromEventStream({ + variables: { + input: { + eventStreamId: sseEventStreamId, + queryId: queryListenerToRemove.queryId, + }, + }, + }); + } + + set(activeQueryListenersState, requiredQueryListeners); + }, + [addQueryToEventStream, removeQueryFromEventStream, sseEventStreamId], + ); + + const debouncedUpdateQueryListeners = useDebouncedCallback( + updateQueryListeners, + 1000, + { leading: true }, + ); + + useEffect(() => { + if (!sseEventStreamId) { + return; + } + + const areRequiredQueryListenersDifferentFromActiveQueryListeners = + compareArraysOfObjectsByProperty( + requiredQueryListeners, + activeQueryListeners, + 'queryId', + ); + + if (areRequiredQueryListenersDifferentFromActiveQueryListeners) { + debouncedUpdateQueryListeners(); + } + }, [ + sseEventStreamId, + requiredQueryListeners, + activeQueryListeners, + debouncedUpdateQueryListeners, + ]); + + return null; +}; diff --git a/packages/twenty-front/src/modules/subscription/contexts/SseClientContext.ts b/packages/twenty-front/src/modules/sse-db-event/contexts/SseClientContext.ts similarity index 100% rename from packages/twenty-front/src/modules/subscription/contexts/SseClientContext.ts rename to packages/twenty-front/src/modules/sse-db-event/contexts/SseClientContext.ts diff --git a/packages/twenty-front/src/modules/sse-db-event/graphql/mutations/AddQueryToEventStreamMutation.ts b/packages/twenty-front/src/modules/sse-db-event/graphql/mutations/AddQueryToEventStreamMutation.ts new file mode 100644 index 00000000000..6eabc19d59c --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/graphql/mutations/AddQueryToEventStreamMutation.ts @@ -0,0 +1,7 @@ +import { gql } from '@apollo/client'; + +export const ADD_QUERY_TO_EVENT_STREAM_MUTATION = gql` + mutation AddQueryToEventStream($input: AddQuerySubscriptionInput!) { + addQueryToEventStream(input: $input) + } +`; diff --git a/packages/twenty-front/src/modules/sse-db-event/graphql/mutations/RemoveQueryFromEventStreamMutation.ts b/packages/twenty-front/src/modules/sse-db-event/graphql/mutations/RemoveQueryFromEventStreamMutation.ts new file mode 100644 index 00000000000..80870af29bd --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/graphql/mutations/RemoveQueryFromEventStreamMutation.ts @@ -0,0 +1,9 @@ +import { gql } from '@apollo/client'; + +export const REMOVE_QUERY_FROM_EVENT_STREAM_MUTATION = gql` + mutation RemoveQueryFromEventStream( + $input: RemoveQueryFromEventStreamInput! + ) { + removeQueryFromEventStream(input: $input) + } +`; diff --git a/packages/twenty-front/src/modules/sse-db-event/graphql/subscriptions/OnEventSubscription.ts b/packages/twenty-front/src/modules/sse-db-event/graphql/subscriptions/OnEventSubscription.ts new file mode 100644 index 00000000000..a9360e8fa2b --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/graphql/subscriptions/OnEventSubscription.ts @@ -0,0 +1,25 @@ +import { gql } from '@apollo/client'; + +export const ON_EVENT_SUBSCRIPTION = gql` + subscription OnEventSubscription($eventStreamId: String!) { + onEventSubscription(eventStreamId: $eventStreamId) { + eventStreamId + eventWithQueryIdsList { + event { + action + objectNameSingular + recordId + userId + workspaceMemberId + properties { + updatedFields + before + after + diff + } + } + queryIds + } + } + } +`; diff --git a/packages/twenty-front/src/modules/subscription/graphql/subscriptions/onDbEvent.ts b/packages/twenty-front/src/modules/sse-db-event/graphql/subscriptions/onDbEvent.ts similarity index 100% rename from packages/twenty-front/src/modules/subscription/graphql/subscriptions/onDbEvent.ts rename to packages/twenty-front/src/modules/sse-db-event/graphql/subscriptions/onDbEvent.ts diff --git a/packages/twenty-front/src/modules/subscription/hooks/__tests__/useOnDbEvent.test.tsx b/packages/twenty-front/src/modules/sse-db-event/hooks/__tests__/useOnDbEvent.test.tsx similarity index 99% rename from packages/twenty-front/src/modules/subscription/hooks/__tests__/useOnDbEvent.test.tsx rename to packages/twenty-front/src/modules/sse-db-event/hooks/__tests__/useOnDbEvent.test.tsx index 399b328c59f..147fbcfc355 100644 --- a/packages/twenty-front/src/modules/subscription/hooks/__tests__/useOnDbEvent.test.tsx +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/__tests__/useOnDbEvent.test.tsx @@ -1,9 +1,9 @@ import { renderHook } from '@testing-library/react'; import { createClient } from 'graphql-sse'; +import { useOnDbEvent } from '@/sse-db-event/hooks/useOnDbEvent'; import { DatabaseEventAction } from '~/generated/graphql'; import { getTokenPair } from '~/modules/apollo/utils/getTokenPair'; -import { useOnDbEvent } from '@/subscription/hooks/useOnDbEvent'; jest.mock('~/modules/apollo/utils/getTokenPair'); jest.mock('graphql-sse'); diff --git a/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToObjectRecordEventsForQuery.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToObjectRecordEventsForQuery.ts new file mode 100644 index 00000000000..52223b53e26 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToObjectRecordEventsForQuery.ts @@ -0,0 +1,76 @@ +import { requiredQueryListenersState } from '@/sse-db-event/states/requiredQueryListenersState'; +import { getObjectRecordEventsForQueryEventName } from '@/sse-db-event/utils/getObjectRecordEventsForQueryEventName'; +import { getSnapshotValue } from '@/ui/utilities/state/utils/getSnapshotValue'; +import { useEffect } from 'react'; +import { useRecoilCallback } from 'recoil'; +import { type RecordGqlOperationSignature } from 'twenty-shared/types'; +import { type ObjectRecordEvent } from '~/generated/graphql'; + +export const useListenToObjectRecordEventsForQuery = ({ + queryId, + operationSignature, + onObjectRecordEvents, +}: { + queryId: string; + operationSignature: RecordGqlOperationSignature; + onObjectRecordEvents: (objectRecordEvents: ObjectRecordEvent[]) => void; +}) => { + useEffect(() => { + const eventName = getObjectRecordEventsForQueryEventName(queryId); + + const handleOnObjectRecordEventsForQuery = (event: Event) => { + const objectRecordEvents = (event as CustomEvent) + .detail; + + onObjectRecordEvents(objectRecordEvents); + }; + + window.addEventListener(eventName, handleOnObjectRecordEventsForQuery); + + return () => { + window.removeEventListener(eventName, handleOnObjectRecordEventsForQuery); + }; + }, [onObjectRecordEvents, queryId]); + + const changeQueryIdListenState = useRecoilCallback( + ({ set, snapshot }) => + (shouldListen: boolean, queryId: string) => { + const currentRequiredQueryListeners = getSnapshotValue( + snapshot, + requiredQueryListenersState, + ); + + const listeningForThisQueryIsActive = + currentRequiredQueryListeners.some( + (listener) => listener.queryId === queryId, + ); + + if (shouldListen === listeningForThisQueryIsActive) { + return; + } + + if (shouldListen) { + set(requiredQueryListenersState, [ + ...currentRequiredQueryListeners, + { queryId, operationSignature }, + ]); + } else { + set( + requiredQueryListenersState, + currentRequiredQueryListeners.filter( + (listener) => listener.queryId !== queryId, + ), + ); + } + }, + [operationSignature], + ); + + useEffect(() => { + changeQueryIdListenState(true, queryId); + + return () => { + changeQueryIdListenState(false, queryId); + }; + }, [changeQueryIdListenState, queryId]); +}; diff --git a/packages/twenty-front/src/modules/subscription/hooks/useOnDbEvent.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useOnDbEvent.ts similarity index 89% rename from packages/twenty-front/src/modules/subscription/hooks/useOnDbEvent.ts rename to packages/twenty-front/src/modules/sse-db-event/hooks/useOnDbEvent.ts index aecd5e62cfa..22acb82e0e6 100644 --- a/packages/twenty-front/src/modules/subscription/hooks/useOnDbEvent.ts +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/useOnDbEvent.ts @@ -1,5 +1,5 @@ -import { ON_DB_EVENT } from '@/subscription/graphql/subscriptions/onDbEvent'; -import { useSseClient } from '@/subscription/hooks/useSseClient.util'; +import { ON_DB_EVENT } from '@/sse-db-event/graphql/subscriptions/onDbEvent'; +import { useSseClient } from '@/sse-db-event/hooks/useSseClient.util'; import { useEffect } from 'react'; import { type Subscription, diff --git a/packages/twenty-front/src/modules/subscription/hooks/useSseClient.util.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useSseClient.util.ts similarity index 100% rename from packages/twenty-front/src/modules/subscription/hooks/useSseClient.util.ts rename to packages/twenty-front/src/modules/sse-db-event/hooks/useSseClient.util.ts diff --git a/packages/twenty-front/src/modules/sse-db-event/states/activeQueryListenersState.ts b/packages/twenty-front/src/modules/sse-db-event/states/activeQueryListenersState.ts new file mode 100644 index 00000000000..3fd1a815ade --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/states/activeQueryListenersState.ts @@ -0,0 +1,9 @@ +import { type RecordGqlOperationSignature } from 'twenty-shared/types'; +import { createState } from 'twenty-ui/utilities'; + +export const activeQueryListenersState = createState< + { queryId: string; operationSignature: RecordGqlOperationSignature }[] +>({ + key: 'activeQueryListenersState', + defaultValue: [], +}); diff --git a/packages/twenty-front/src/modules/sse-db-event/states/requiredQueryListenersState.ts b/packages/twenty-front/src/modules/sse-db-event/states/requiredQueryListenersState.ts new file mode 100644 index 00000000000..d45e64b07e2 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/states/requiredQueryListenersState.ts @@ -0,0 +1,9 @@ +import { type RecordGqlOperationSignature } from 'twenty-shared/types'; +import { createState } from 'twenty-ui/utilities'; + +export const requiredQueryListenersState = createState< + { queryId: string; operationSignature: RecordGqlOperationSignature }[] +>({ + key: 'requiredQueryListenersState', + defaultValue: [], +}); diff --git a/packages/twenty-front/src/modules/sse-db-event/states/sseEventStreamIdState.ts b/packages/twenty-front/src/modules/sse-db-event/states/sseEventStreamIdState.ts new file mode 100644 index 00000000000..ef7138a4af1 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/states/sseEventStreamIdState.ts @@ -0,0 +1,6 @@ +import { createState } from 'twenty-ui/utilities'; + +export const sseEventStreamIdState = createState({ + key: 'sseEventStreamIdState', + defaultValue: null, +}); diff --git a/packages/twenty-front/src/modules/sse-db-event/types/ObjectRecordEventsByQueryId.ts b/packages/twenty-front/src/modules/sse-db-event/types/ObjectRecordEventsByQueryId.ts new file mode 100644 index 00000000000..a81891ce09f --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/types/ObjectRecordEventsByQueryId.ts @@ -0,0 +1,3 @@ +import { type ObjectRecordEvent } from '~/generated/graphql'; + +export type ObjectRecordEventsByQueryId = Record; diff --git a/packages/twenty-front/src/modules/sse-db-event/utils/dispatchObjectRecordEventsWithQueryIds.ts b/packages/twenty-front/src/modules/sse-db-event/utils/dispatchObjectRecordEventsWithQueryIds.ts new file mode 100644 index 00000000000..bdcb4edb593 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/utils/dispatchObjectRecordEventsWithQueryIds.ts @@ -0,0 +1,30 @@ +import { type ObjectRecordEventsByQueryId } from '@/sse-db-event/types/ObjectRecordEventsByQueryId'; +import { getObjectRecordEventsForQueryEventName } from '@/sse-db-event/utils/getObjectRecordEventsForQueryEventName'; +import { isDefined } from 'twenty-shared/utils'; +import { type EventWithQueryIds } from '~/generated/graphql'; + +export const dispatchObjectRecordEventsWithQueryIds = ( + objectRecordEventsWithQueryIds: EventWithQueryIds[], +) => { + const objectRecordEventsByQueryId: ObjectRecordEventsByQueryId = {}; + + for (const objectRecordEventWithQueryIds of objectRecordEventsWithQueryIds) { + for (const queryId of objectRecordEventWithQueryIds.queryIds) { + if (!isDefined(objectRecordEventsByQueryId[queryId])) { + objectRecordEventsByQueryId[queryId] = []; + } + + objectRecordEventsByQueryId[queryId].push( + objectRecordEventWithQueryIds.event, + ); + } + } + + for (const queryId in objectRecordEventsByQueryId) { + window.dispatchEvent( + new CustomEvent(getObjectRecordEventsForQueryEventName(queryId), { + detail: objectRecordEventsByQueryId[queryId], + }), + ); + } +}; diff --git a/packages/twenty-front/src/modules/subscription/utils/getDefaultRecordFieldsToListen.util.ts b/packages/twenty-front/src/modules/sse-db-event/utils/getDefaultRecordFieldsToListen.ts similarity index 100% rename from packages/twenty-front/src/modules/subscription/utils/getDefaultRecordFieldsToListen.util.ts rename to packages/twenty-front/src/modules/sse-db-event/utils/getDefaultRecordFieldsToListen.ts diff --git a/packages/twenty-front/src/modules/sse-db-event/utils/getObjectRecordEventsForQueryEventName.ts b/packages/twenty-front/src/modules/sse-db-event/utils/getObjectRecordEventsForQueryEventName.ts new file mode 100644 index 00000000000..eba06294d02 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/utils/getObjectRecordEventsForQueryEventName.ts @@ -0,0 +1,3 @@ +export const getObjectRecordEventsForQueryEventName = (queryId: string) => { + return `object-record-events-${queryId}`; +}; diff --git a/packages/twenty-front/src/modules/subscription/components/SubscriptionProvider.tsx b/packages/twenty-front/src/modules/subscription/components/SubscriptionProvider.tsx deleted file mode 100644 index 78171da6ca6..00000000000 --- a/packages/twenty-front/src/modules/subscription/components/SubscriptionProvider.tsx +++ /dev/null @@ -1,21 +0,0 @@ -import { SubscriptionProviderEffect } from '@/subscription/components/SubscriptionProviderEffect'; -import { SseClientContext } from '@/subscription/contexts/SseClientContext'; -import { useSseClient } from '@/subscription/hooks/useSseClient.util'; -import { type ReactNode } from 'react'; - -type SubscriptionProviderProps = { - children: ReactNode; -}; - -export const SubscriptionProvider = ({ - children, -}: SubscriptionProviderProps) => { - const { sseClient } = useSseClient(); - - return ( - - - {children} - - ); -}; diff --git a/packages/twenty-front/src/modules/subscription/components/SubscriptionProviderEffect.tsx b/packages/twenty-front/src/modules/subscription/components/SubscriptionProviderEffect.tsx deleted file mode 100644 index 7e5f842de4f..00000000000 --- a/packages/twenty-front/src/modules/subscription/components/SubscriptionProviderEffect.tsx +++ /dev/null @@ -1,68 +0,0 @@ -import { ON_SUBSCRIPTION_MATCH } from '@/subscription/graphql/subscriptions/onSubscriptionMatch'; -import { useSseClient } from '@/subscription/hooks/useSseClient.util'; -import { subscriptionRegistryState } from '@/subscription/states/subscriptionRegistryState'; -import { print } from 'graphql'; -import { useEffect, useMemo } from 'react'; -import { useRecoilValue } from 'recoil'; -import { isDefined } from 'twenty-shared/utils'; -import { type SubscriptionMatches } from '~/generated/graphql'; - -export const SubscriptionProviderEffect = () => { - const subscriptionRegistry = useRecoilValue(subscriptionRegistryState); - - const { sseClient } = useSseClient(); - - const subscriptions = useMemo(() => { - return Array.from(subscriptionRegistry.values()).map((entry) => ({ - id: entry.id, - query: entry.query, - })); - }, [subscriptionRegistry]); - - useEffect(() => { - if (!sseClient || subscriptions.length === 0) { - return; - } - - const unsubscribe = sseClient.subscribe( - { - query: print(ON_SUBSCRIPTION_MATCH), - variables: { subscriptions }, - }, - { - next: (value) => { - const data = value.data as { - onSubscriptionMatch: SubscriptionMatches; - } | null; - - if (!data?.onSubscriptionMatch?.matches) { - return; - } - - for (const match of data.onSubscriptionMatch.matches) { - for (const subscriptionId of match.subscriptionIds) { - const entry = subscriptionRegistry.get(subscriptionId); - - if (!isDefined(entry)) { - continue; - } - - entry.onRefetch(); - } - } - }, - error: (error) => { - // eslint-disable-next-line no-console - console.error('Subscription error:', error); - }, - complete: () => {}, - }, - ); - - return () => { - unsubscribe(); - }; - }, [sseClient, subscriptions, subscriptionRegistry]); - - return null; -}; diff --git a/packages/twenty-front/src/modules/subscription/graphql/subscriptions/onSubscriptionMatch.ts b/packages/twenty-front/src/modules/subscription/graphql/subscriptions/onSubscriptionMatch.ts deleted file mode 100644 index 98a968f92ab..00000000000 --- a/packages/twenty-front/src/modules/subscription/graphql/subscriptions/onSubscriptionMatch.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { gql } from '@apollo/client'; - -export const ON_SUBSCRIPTION_MATCH = gql` - subscription OnSubscriptionMatch($subscriptions: [SubscriptionInput!]!) { - onSubscriptionMatch(subscriptions: $subscriptions) { - matches { - subscriptionIds - event { - action - objectNameSingular - eventDate - record - updatedFields - } - } - } - } -`; diff --git a/packages/twenty-front/src/modules/subscription/hooks/useSubscribeToRefetch.ts b/packages/twenty-front/src/modules/subscription/hooks/useSubscribeToRefetch.ts deleted file mode 100644 index beb52d69520..00000000000 --- a/packages/twenty-front/src/modules/subscription/hooks/useSubscribeToRefetch.ts +++ /dev/null @@ -1,50 +0,0 @@ -import { subscriptionRegistryState } from '@/subscription/states/subscriptionRegistryState'; -import { type DocumentNode, print } from 'graphql'; -import { useEffect, useId } from 'react'; -import { useSetRecoilState } from 'recoil'; - -type UseSubscribeToRefetchParams = { - query: DocumentNode; - variables?: Record; - refetch: () => void; - skip?: boolean; -}; - -export const useSubscribeToRefetch = ({ - query, - variables, - refetch, - skip = false, -}: UseSubscribeToRefetchParams) => { - const setRegistry = useSetRecoilState(subscriptionRegistryState); - const subscriptionId = useId(); - - const queryString = JSON.stringify({ - query: print(query), - variables, - }); - - useEffect(() => { - if (skip) { - return; - } - - setRegistry((prev) => { - const next = new Map(prev); - next.set(subscriptionId, { - id: subscriptionId, - query: queryString, - onRefetch: refetch, - }); - return next; - }); - - return () => { - setRegistry((prev) => { - const next = new Map(prev); - next.delete(subscriptionId); - return next; - }); - }; - }, [subscriptionId, queryString, refetch, skip, setRegistry]); -}; diff --git a/packages/twenty-front/src/modules/subscription/states/subscriptionRegistryState.ts b/packages/twenty-front/src/modules/subscription/states/subscriptionRegistryState.ts deleted file mode 100644 index cc5e2d15333..00000000000 --- a/packages/twenty-front/src/modules/subscription/states/subscriptionRegistryState.ts +++ /dev/null @@ -1,14 +0,0 @@ -import { createState } from 'twenty-ui/utilities'; - -export type SubscriptionEntry = { - id: string; - query: string; - onRefetch: () => void; -}; - -export const subscriptionRegistryState = createState< - Map ->({ - key: 'subscriptionRegistryState', - defaultValue: new Map(), -}); diff --git a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunCard.tsx b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunCard.tsx index cd2653a8336..d07eab4cf72 100644 --- a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunCard.tsx +++ b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunCard.tsx @@ -1,5 +1,5 @@ import { SKELETON_LOADER_HEIGHT_SIZES } from '@/activities/components/SkeletonLoader'; -import { ListenRecordUpdatesEffect } from '@/subscription/components/ListenRecordUpdatesEffect'; +import { ListenRecordUpdatesEffect } from '@/sse-db-event/components/ListenRecordUpdatesEffect'; import { useTargetRecord } from '@/ui/layout/contexts/useTargetRecord'; import { getWorkflowVisualizerComponentInstanceId } from '@/workflow/utils/getWorkflowVisualizerComponentInstanceId'; import { WorkflowRunVisualizer } from '@/workflow/workflow-diagram/components/WorkflowRunVisualizer'; diff --git a/packages/twenty-server/src/engine/subscriptions/dtos/object-record-event.dto.ts b/packages/twenty-server/src/engine/subscriptions/dtos/object-record-event.dto.ts index d19e95fcd17..0f5acc64eaf 100644 --- a/packages/twenty-server/src/engine/subscriptions/dtos/object-record-event.dto.ts +++ b/packages/twenty-server/src/engine/subscriptions/dtos/object-record-event.dto.ts @@ -1,9 +1,13 @@ import { Field, ObjectType } from '@nestjs/graphql'; +import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { ObjectRecordEventPropertiesDTO } from 'src/engine/subscriptions/dtos/object-record-event-properties.dto'; @ObjectType('ObjectRecordEvent') export class ObjectRecordEventDTO { + @Field(() => DatabaseEventAction) + action: DatabaseEventAction; + @Field(() => String) objectNameSingular: string; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts index bcfe672e479..cb323690061 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts @@ -23,6 +23,7 @@ import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription import { EventStreamService } from 'src/engine/subscriptions/event-stream.service'; import { SubscriptionService } from 'src/engine/subscriptions/subscription.service'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; +import { parseEventNameOrThrow } from 'src/engine/workspace-event-emitter/utils/parse-event-name'; @Resolver() @UseGuards(WorkspaceAuthGuard, UserAuthGuard, NoPermissionGuard) @@ -144,9 +145,16 @@ export class WorkspaceEventEmitterResolver { }[] = []; for (const event of payload.workspaceEventBatch.events) { + const eventName = parseEventNameOrThrow( + payload.workspaceEventBatch.name, + ); + + const action = eventName.action; + const eventWithObjectName = { - objectNameSingular, ...event, + objectNameSingular, + action, }; const matchedQueryIds = diff --git a/packages/twenty-shared/src/utils/array/__tests__/compareArraysOfObjectsByProperty.test.ts b/packages/twenty-shared/src/utils/array/__tests__/compareArraysOfObjectsByProperty.test.ts new file mode 100644 index 00000000000..cc949b6e4e1 --- /dev/null +++ b/packages/twenty-shared/src/utils/array/__tests__/compareArraysOfObjectsByProperty.test.ts @@ -0,0 +1,92 @@ +import { compareArraysOfObjectsByProperty } from '@/utils/array/compareArraysOfObjectsByProperty'; + +type TestObject = { + id: string; + name: string; +}; + +describe('compareArraysOfObjectsByProperty', () => { + it('should return false when both arrays are empty', () => { + expect(compareArraysOfObjectsByProperty([], [], 'id')).toBe(false); + }); + + it('should return false when arrays have same objects by property', () => { + const arrayA: TestObject[] = [ + { id: '1', name: 'Test 1' }, + { id: '2', name: 'Test 2' }, + ]; + const arrayB: TestObject[] = [ + { id: '1', name: 'Different Name' }, + { id: '2', name: 'Another Name' }, + ]; + + expect(compareArraysOfObjectsByProperty(arrayA, arrayB, 'id')).toBe(false); + }); + + it('should return true when arrays have different lengths', () => { + const arrayA: TestObject[] = [ + { id: '1', name: 'Test 1' }, + { id: '2', name: 'Test 2' }, + ]; + const arrayB: TestObject[] = [{ id: '1', name: 'Test 1' }]; + + expect(compareArraysOfObjectsByProperty(arrayA, arrayB, 'id')).toBe(true); + }); + + it('should return true when arrayA has items not in arrayB', () => { + const arrayA: TestObject[] = [ + { id: '1', name: 'Test 1' }, + { id: '2', name: 'Test 2' }, + ]; + const arrayB: TestObject[] = [ + { id: '1', name: 'Test 1' }, + { id: '3', name: 'Test 3' }, + ]; + + expect(compareArraysOfObjectsByProperty(arrayA, arrayB, 'id')).toBe(true); + }); + + it('should return true when arrayB has items not in arrayA', () => { + const arrayA: TestObject[] = [ + { id: '1', name: 'Test 1' }, + { id: '3', name: 'Test 3' }, + ]; + const arrayB: TestObject[] = [ + { id: '1', name: 'Test 1' }, + { id: '2', name: 'Test 2' }, + ]; + + expect(compareArraysOfObjectsByProperty(arrayA, arrayB, 'id')).toBe(true); + }); + + it('should return false when arrays have same items in different order', () => { + const arrayA: TestObject[] = [ + { id: '1', name: 'Test 1' }, + { id: '2', name: 'Test 2' }, + { id: '3', name: 'Test 3' }, + ]; + const arrayB: TestObject[] = [ + { id: '3', name: 'Test 3' }, + { id: '1', name: 'Test 1' }, + { id: '2', name: 'Test 2' }, + ]; + + expect(compareArraysOfObjectsByProperty(arrayA, arrayB, 'id')).toBe(false); + }); + + it('should compare by the specified property', () => { + const arrayA: TestObject[] = [ + { id: '1', name: 'Alpha' }, + { id: '2', name: 'Beta' }, + ]; + const arrayB: TestObject[] = [ + { id: '3', name: 'Alpha' }, + { id: '4', name: 'Beta' }, + ]; + + expect(compareArraysOfObjectsByProperty(arrayA, arrayB, 'name')).toBe( + false, + ); + expect(compareArraysOfObjectsByProperty(arrayA, arrayB, 'id')).toBe(true); + }); +}); diff --git a/packages/twenty-shared/src/utils/array/compareArraysOfObjectsByProperty.ts b/packages/twenty-shared/src/utils/array/compareArraysOfObjectsByProperty.ts new file mode 100644 index 00000000000..df6f0a10660 --- /dev/null +++ b/packages/twenty-shared/src/utils/array/compareArraysOfObjectsByProperty.ts @@ -0,0 +1,15 @@ +export const compareArraysOfObjectsByProperty = ( + arrayA: T[], + arrayB: T[], + property: K, +) => { + return ( + arrayA.length !== arrayB.length || + arrayA.some( + (item) => !arrayB.some((itemB) => itemB[property] === item[property]), + ) || + arrayB.some( + (item) => !arrayA.some((itemA) => itemA[property] === item[property]), + ) + ); +}; diff --git a/packages/twenty-shared/src/utils/index.ts b/packages/twenty-shared/src/utils/index.ts index 6f0a7964f68..dd343f9036c 100644 --- a/packages/twenty-shared/src/utils/index.ts +++ b/packages/twenty-shared/src/utils/index.ts @@ -8,6 +8,7 @@ */ export { applyDiff } from './applyDiff'; +export { compareArraysOfObjectsByProperty } from './array/compareArraysOfObjectsByProperty'; export { filterOutByProperty } from './array/filterOutByProperty'; export { findById } from './array/findById'; export { findByProperty } from './array/findByProperty';