diff --git a/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx b/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx index f47de6b6f2..59583df0f2 100644 --- a/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx @@ -1,6 +1,7 @@ -import { useContext } from 'react'; +import { useCallback, useContext } from 'react'; import { RecordBoardContext } from '@/object-record/record-board/contexts/RecordBoardContext'; +import { useTriggerRecordBoardInitialQuery } from '@/object-record/record-board/hooks/useTriggerRecordBoardInitialQuery'; import { useRecordIndexContextOrThrow } from '@/object-record/record-index/contexts/RecordIndexContext'; import { useRecordIndexGroupCommonQueryVariables } from '@/object-record/record-index/hooks/useRecordIndexGroupCommonQueryVariables'; import { useListenToEventsForQuery } from '@/sse-db-event/hooks/useListenToEventsForQuery'; @@ -10,9 +11,15 @@ export const RecordBoardSSESubscribeEffect = () => { const { objectMetadataItem } = useRecordIndexContextOrThrow(); const { combinedFilters, orderBy } = useRecordIndexGroupCommonQueryVariables(); + const { triggerRecordBoardInitialQuery } = + useTriggerRecordBoardInitialQuery(); const queryId = `record-board-${recordBoardId}`; + const reloadBoard = useCallback(() => { + triggerRecordBoardInitialQuery({ shouldResetScroll: false }); + }, [triggerRecordBoardInitialQuery]); + useListenToEventsForQuery({ queryId, operationSignature: { @@ -22,6 +29,7 @@ export const RecordBoardSSESubscribeEffect = () => { orderBy, }, }, + onSseReconnected: reloadBoard, }); return null; diff --git a/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx index a128ade281..97a7928c12 100644 --- a/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx @@ -1,12 +1,15 @@ -import { useMemo } from 'react'; +import { useCallback, useMemo } from 'react'; import { flattenedFieldMetadataItemsSelector } from '@/object-metadata/states/flattenedFieldMetadataItemsSelector'; import { turnSortsIntoOrderBy } from '@/object-record/object-sort-dropdown/utils/turnSortsIntoOrderBy'; 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 { useRefetchAggregateQueriesForObjectMetadataItem } from '@/object-record/hooks/useRefetchAggregateQueriesForObjectMetadataItem'; import { useRecordIndexContextOrThrow } from '@/object-record/record-index/contexts/RecordIndexContext'; import { currentRecordSortsComponentState } from '@/object-record/record-sort/states/currentRecordSortsComponentState'; +import { useRecordTableContextOrThrow } from '@/object-record/record-table/contexts/RecordTableContext'; +import { useResetVirtualizationBecauseDataChanged } from '@/object-record/record-table/virtualization/hooks/useResetVirtualizationBecauseDataChanged'; import { useListenToEventsForQuery } from '@/sse-db-event/hooks/useListenToEventsForQuery'; import { useAtomComponentStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomComponentStateValue'; import { useAtomStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomStateValue'; @@ -14,8 +17,26 @@ import { computeRecordGqlOperationFilter } from 'twenty-shared/utils'; export const RecordTableVirtualizedSSESubscribeEffect = () => { const { objectMetadataItem } = useRecordIndexContextOrThrow(); + const { objectNameSingular } = useRecordTableContextOrThrow(); const { filterValueDependencies } = useFilterValueDependencies(); + const { resetVirtualizationBecauseDataChanged } = + useResetVirtualizationBecauseDataChanged(objectNameSingular); + + const { refetchAggregateQueriesForObjectMetadataItem } = + useRefetchAggregateQueriesForObjectMetadataItem(); + + const reloadTable = useCallback(async () => { + await Promise.all([ + resetVirtualizationBecauseDataChanged(), + refetchAggregateQueriesForObjectMetadataItem({ objectMetadataItem }), + ]); + }, [ + objectMetadataItem, + refetchAggregateQueriesForObjectMetadataItem, + resetVirtualizationBecauseDataChanged, + ]); + const flattenedFieldMetadataItems = useAtomStateValue( flattenedFieldMetadataItemsSelector, ); @@ -60,6 +81,7 @@ export const RecordTableVirtualizedSSESubscribeEffect = () => { useListenToEventsForQuery({ queryId, operationSignature, + onSseReconnected: reloadTable, }); return null; diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx index c19105ab36..edefa2dbdd 100644 --- a/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx @@ -1,8 +1,10 @@ import { useHasAccessTokenPair } from '@/auth/hooks/useHasAccessTokenPair'; import { tokenPairState } from '@/auth/states/tokenPairState'; +import { useListenToBrowserEvent } from '@/browser-event/hooks/useListenToBrowserEvent'; import { dispatchBrowserEvent } from '@/browser-event/utils/dispatchBrowserEvent'; import { useResyncMetadataStore } from '@/metadata-store/hooks/useResyncMetadataStore'; import { SSE_CLIENT_RECONNECTED_EVENT_NAME } from '@/sse-db-event/constants/SseClientReconnectedEventName'; +import { SSE_RESYNC_DEBOUNCE_TIME_IN_MS } from '@/sse-db-event/constants/SseResyncDebounceTimeInMs'; import { useHandleSseClientConnectionRetry } from '@/sse-db-event/hooks/useHandleSseClientConnectionRetry'; import { activeQueryListenersState } from '@/sse-db-event/states/activeQueryListenersState'; import { sseClientState } from '@/sse-db-event/states/sseClientState'; @@ -12,6 +14,7 @@ import { isNonEmptyArray } from '@sniptt/guards'; import { createClient } from 'graphql-sse'; import { useCallback, useEffect } from 'react'; import { isDefined } from 'twenty-shared/utils'; +import { useDebouncedCallback } from 'use-debounce'; import { REACT_APP_SERVER_BASE_URL } from '~/config'; import { useStore } from 'jotai'; @@ -22,6 +25,17 @@ export const SSEClientEffect = () => { const tokenPair = useAtomStateValue(tokenPairState); const { resyncMetadataStore } = useResyncMetadataStore(); + const debouncedResyncMetadataStore = useDebouncedCallback( + resyncMetadataStore, + SSE_RESYNC_DEBOUNCE_TIME_IN_MS, + { leading: false }, + ); + + useListenToBrowserEvent({ + eventName: SSE_CLIENT_RECONNECTED_EVENT_NAME, + onBrowserEvent: debouncedResyncMetadataStore, + }); + const handleSSEClientConnected = useCallback( (reconnected: boolean) => { const currentActiveQueryListeners = store.get( @@ -33,11 +47,10 @@ export const SSEClientEffect = () => { } if (reconnected) { - resyncMetadataStore(); dispatchBrowserEvent(SSE_CLIENT_RECONNECTED_EVENT_NAME); } }, - [store, resyncMetadataStore], + [store], ); const { handleSseClientConnectionRetry } = 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 index 3f5e046ee0..00693210ed 100644 --- a/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx @@ -7,7 +7,6 @@ import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdSta import { sseEventStreamReadyState } from '@/sse-db-event/states/sseEventStreamReadyState'; import { isGracefullyHandledEventStreamError } from '@/sse-db-event/utils/isGracefullyHandledEventStreamError'; import { useAtomStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomStateValue'; -import { CombinedGraphQLErrors } from '@apollo/client/errors'; import { captureException } from '@sentry/react'; import { useMutation } from '@apollo/client/react'; import { isNonEmptyString } from '@sniptt/guards'; @@ -44,25 +43,26 @@ export const SSEQuerySubscribeEffect = () => { const handleError = useCallback( (error: unknown) => { - if (CombinedGraphQLErrors.is(error)) { - const extensions = getGraphqlErrorExtensionsFromError(error); + const extensions = getGraphqlErrorExtensionsFromError(error); - if ( - !isGracefullyHandledEventStreamError({ - subCode: extensions?.subCode, - code: extensions?.code, - }) - ) { - captureException( - new Error(`Unhandled error for event stream: ${error.message}`, { - cause: error, - }), - ); - } - - store.set(activeQueryListenersState.atom, []); - store.set(shouldDestroyEventStreamState.atom, true); + if ( + !isGracefullyHandledEventStreamError({ + subCode: extensions?.subCode, + code: extensions?.code, + }) + ) { + captureException( + new Error( + `Unhandled error for event stream: ${ + error instanceof Error ? error.message : String(error) + }`, + { cause: error }, + ), + ); } + + store.set(activeQueryListenersState.atom, []); + store.set(shouldDestroyEventStreamState.atom, true); }, [store], ); diff --git a/packages/twenty-front/src/modules/sse-db-event/constants/SseResyncDebounceTimeInMs.ts b/packages/twenty-front/src/modules/sse-db-event/constants/SseResyncDebounceTimeInMs.ts new file mode 100644 index 0000000000..93f0244889 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/constants/SseResyncDebounceTimeInMs.ts @@ -0,0 +1 @@ +export const SSE_RESYNC_DEBOUNCE_TIME_IN_MS = 1_000; diff --git a/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts index 399bb2b605..00648be15c 100644 --- a/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts @@ -1,20 +1,28 @@ +import { useListenToBrowserEvent } from '@/browser-event/hooks/useListenToBrowserEvent'; +import { SSE_CLIENT_RECONNECTED_EVENT_NAME } from '@/sse-db-event/constants/SseClientReconnectedEventName'; +import { SSE_RESYNC_DEBOUNCE_TIME_IN_MS } from '@/sse-db-event/constants/SseResyncDebounceTimeInMs'; import { useChangeQueryListenState } from '@/sse-db-event/hooks/useChangeQueryListenState'; -import { useEffect } from 'react'; +import { captureException } from '@sentry/react'; +import { useCallback, useEffect } from 'react'; import { type MetadataGqlOperationSignature, type RecordGqlOperationSignature, } from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { useDebouncedCallback } from 'use-debounce'; export const useListenToEventsForQuery = ({ queryId, operationSignature, skip = false, + onSseReconnected, }: { queryId: string; operationSignature: | RecordGqlOperationSignature | MetadataGqlOperationSignature; skip?: boolean; + onSseReconnected?: () => void | Promise; }) => { const { changeQueryIdListenState } = useChangeQueryListenState(); @@ -29,4 +37,35 @@ export const useListenToEventsForQuery = ({ changeQueryIdListenState(false, queryId, operationSignature); }; }, [changeQueryIdListenState, queryId, operationSignature, skip]); + + const handleSseReconnected = useCallback(() => { + if (skip || !isDefined(onSseReconnected)) { + return; + } + + const captureResyncError = (error: unknown) => { + captureException( + new Error(`Failed to resync "${queryId}" after SSE reconnection`, { + cause: error, + }), + ); + }; + + try { + void Promise.resolve(onSseReconnected()).catch(captureResyncError); + } catch (error) { + captureResyncError(error); + } + }, [onSseReconnected, queryId, skip]); + + const debouncedHandleSseReconnected = useDebouncedCallback( + handleSseReconnected, + SSE_RESYNC_DEBOUNCE_TIME_IN_MS, + { leading: false }, + ); + + useListenToBrowserEvent({ + eventName: SSE_CLIENT_RECONNECTED_EVENT_NAME, + onBrowserEvent: debouncedHandleSseReconnected, + }); }; diff --git a/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts index d15e0a0d92..09bb58a46b 100644 --- a/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts @@ -1,3 +1,5 @@ +import { dispatchBrowserEvent } from '@/browser-event/utils/dispatchBrowserEvent'; +import { SSE_CLIENT_RECONNECTED_EVENT_NAME } from '@/sse-db-event/constants/SseClientReconnectedEventName'; import { ON_EVENT_SUBSCRIPTION } from '@/sse-db-event/graphql/subscriptions/OnEventSubscription'; import { useDispatchMetadataEventsFromSseToBrowserEvents } from '@/sse-db-event/hooks/useDispatchMetadataEventsFromSseToBrowserEvents'; import { useDispatchObjectRecordEventsFromSseToBrowserEvents } from '@/sse-db-event/hooks/useDispatchObjectRecordEventsFromSseToBrowserEvents'; @@ -67,8 +69,25 @@ export const useTriggerEventStreamCreation = () => { store.set(sseEventStreamIdState.atom, newSseEventStreamId); store.set(sseEventStreamReadyState.atom, false); + const isRecreatedEventStream = isDefined( + store.get(lastSseEventReceivedTimestampState.atom), + ); + let hasReceivedFirstEvent = false; + const handleFirstEventReceived = () => { + if (hasReceivedFirstEvent) { + return; + } + + hasReceivedFirstEvent = true; + store.set(sseEventStreamReadyState.atom, true); + + if (isRecreatedEventStream) { + dispatchBrowserEvent(SSE_CLIENT_RECONNECTED_EVENT_NAME); + } + }; + const dispose = sseClient.subscribe( { query: print(ON_EVENT_SUBSCRIPTION), @@ -107,10 +126,7 @@ export const useTriggerEventStreamCreation = () => { return; } - if (!hasReceivedFirstEvent) { - hasReceivedFirstEvent = true; - store.set(sseEventStreamReadyState.atom, true); - } + handleFirstEventReceived(); const eventSubscription = value?.data?.onEventSubscription; @@ -169,10 +185,7 @@ export const useTriggerEventStreamCreation = () => { store.set(shouldDestroyEventStreamState.atom, true); } else { - if (!hasReceivedFirstEvent) { - hasReceivedFirstEvent = true; - store.set(sseEventStreamReadyState.atom, true); - } + handleFirstEventReceived(); const objectRecordEventsWithQueryIds = result?.data?.onEventSubscription diff --git a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx index 13e48fc304..4b0d969c75 100644 --- a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx @@ -24,6 +24,10 @@ export const WorkflowSSESubscribeEffect = ({ objectNameSingular: CoreObjectNameSingular.WorkflowVersion, }); + const requestWorkflowRefetch = useCallback(() => { + setShouldWorkflowRefetchRequest(true); + }, [setShouldWorkflowRefetchRequest]); + useListenToEventsForQuery({ queryId, operationSignature: { @@ -34,14 +38,11 @@ export const WorkflowSSESubscribeEffect = ({ }, }, }, + onSseReconnected: requestWorkflowRefetch, }); - const handleWorkflowVersionCreateOne = useCallback(() => { - setShouldWorkflowRefetchRequest(true); - }, [setShouldWorkflowRefetchRequest]); - useListenToObjectRecordOperationBrowserEvent({ - onObjectRecordOperationBrowserEvent: handleWorkflowVersionCreateOne, + onObjectRecordOperationBrowserEvent: requestWorkflowRefetch, objectMetadataItemId: workflowVersionMetadataItem.id, operationTypes: ['create-one'], });