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 a308c5adab..3627162403 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,7 +1,6 @@ -import { getTokenPair } from '@/apollo/utils/getTokenPair'; import { useIsLogged } from '@/auth/hooks/useIsLogged'; -import { SSE_CONNECTION_RETRY_MAX_WAIT_TIME_IN_MS } from '@/sse-db-event/constants/SseConnectionRetryMaxWaitTimeInMs'; -import { SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_FOR_DEV_MODE } from '@/sse-db-event/constants/SseConnectionRetryWaitTimeInMsForDevMode'; +import { tokenPairState } from '@/auth/states/tokenPairState'; +import { useHandleSseClientConnectionRetry } from '@/sse-db-event/hooks/useHandleSseClientConnectionRetry'; import { activeQueryListenersState } from '@/sse-db-event/states/activeQueryListenersState'; import { sseClientState } from '@/sse-db-event/states/sseClientState'; import { getSnapshotValue } from '@/ui/utilities/state/utils/getSnapshotValue'; @@ -11,13 +10,11 @@ import { useEffect } from 'react'; import { useRecoilCallback, useRecoilState } from 'recoil'; import { isDefined } from 'twenty-shared/utils'; import { REACT_APP_SERVER_BASE_URL } from '~/config'; -import { getIsDevelopmentEnvironment } from '~/utils/getIsDevelopmentEnvironment'; - -import { sleep } from '~/utils/sleep'; export const SSEClientEffect = () => { const isLoggedIn = useIsLogged(); const [sseClient, setSseClient] = useRecoilState(sseClientState); + const [tokenPair] = useRecoilState(tokenPairState); const handleSSEClientConnected = useRecoilCallback( ({ snapshot, set }) => @@ -34,9 +31,11 @@ export const SSEClientEffect = () => { [], ); + const { handleSseClientConnectionRetry } = + useHandleSseClientConnectionRetry(); + useEffect(() => { - if (isLoggedIn && !isDefined(sseClient)) { - const tokenPair = getTokenPair(); + if (isLoggedIn && !isDefined(sseClient) && isDefined(tokenPair)) { const token = tokenPair?.accessOrWorkspaceAgnosticToken?.token; const newSseClient = createClient({ @@ -48,24 +47,20 @@ export const SSEClientEffect = () => { connected: handleSSEClientConnected, }, retryAttempts: Infinity, - retry: async () => { - const randomWaitTimeInMsToSpaceAllClientsReconnection = Math.round( - Math.random() * SSE_CONNECTION_RETRY_MAX_WAIT_TIME_IN_MS, - ); - - const isDevelopmentEnvironment = getIsDevelopmentEnvironment(); - - const waitTimeInMs = isDevelopmentEnvironment - ? SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_FOR_DEV_MODE - : randomWaitTimeInMsToSpaceAllClientsReconnection; - - await sleep(waitTimeInMs); - }, + retry: (retryCount: number) => + handleSseClientConnectionRetry(retryCount, token), }); setSseClient(newSseClient); } - }, [handleSSEClientConnected, isLoggedIn, setSseClient, sseClient]); + }, [ + handleSSEClientConnected, + isLoggedIn, + setSseClient, + sseClient, + tokenPair, + handleSseClientConnectionRetry, + ]); return null; }; diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEEventStreamEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEEventStreamEffect.tsx index 4f878ed91b..7b9b702a03 100644 --- a/packages/twenty-front/src/modules/sse-db-event/components/SSEEventStreamEffect.tsx +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEEventStreamEffect.tsx @@ -6,7 +6,6 @@ import { useTriggerEventStreamDestroy } from '@/sse-db-event/hooks/useTriggerEve import { isCreatingSseEventStreamState } from '@/sse-db-event/states/isCreatingSseEventStreamState'; import { isDestroyingEventStreamState } from '@/sse-db-event/states/isDestroyingEventStreamState'; import { shouldDestroyEventStreamState } from '@/sse-db-event/states/shouldDestroyEventStreamState'; -import { sseClientState } from '@/sse-db-event/states/sseClientState'; import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdState'; import { useIsFeatureEnabled } from '@/workspace/hooks/useIsFeatureEnabled'; import { isNonEmptyArray } from '@apollo/client/utilities'; @@ -17,7 +16,6 @@ import { isDefined } from 'twenty-shared/utils'; import { FeatureFlagKey, OnboardingStatus } from '~/generated/graphql'; export const SSEEventStreamEffect = () => { - const sseClient = useRecoilValue(sseClientState); const objectMetadataItems = useRecoilValue(objectMetadataItemsState); const sseEventStreamId = useRecoilValue(sseEventStreamIdState); @@ -40,9 +38,7 @@ export const SSEEventStreamEffect = () => { useEffect(() => { const isSseClientAvailabble = - isDefined(sseClient) && - !isCreatingSseEventStream && - !isDestroyingEventStream; + !isCreatingSseEventStream && !isDestroyingEventStream; const willCreateEventStream = isSseClientAvailabble && @@ -62,7 +58,7 @@ export const SSEEventStreamEffect = () => { if (willDestroyEventStream) { triggerEventStreamDestroy(); } else if (willCreateEventStream) { - triggerEventStreamCreation(sseClient); + triggerEventStreamCreation(); } }, [ isCreatingSseEventStream, @@ -72,7 +68,6 @@ export const SSEEventStreamEffect = () => { isSseDbEventsEnabled, isDestroyingEventStream, triggerEventStreamDestroy, - sseClient, shouldDestroyEventStream, sseEventStreamId, objectMetadataItems, diff --git a/packages/twenty-front/src/modules/sse-db-event/constants/SseConnectionRetryWaitTimeInMsToAvoidRaceConditions.ts b/packages/twenty-front/src/modules/sse-db-event/constants/SseConnectionRetryWaitTimeInMsToAvoidRaceConditions.ts new file mode 100644 index 0000000000..2a790849d9 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/constants/SseConnectionRetryWaitTimeInMsToAvoidRaceConditions.ts @@ -0,0 +1 @@ +export const SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_TO_AVOID_RACE_CONDITIONS = 1_000; diff --git a/packages/twenty-front/src/modules/sse-db-event/hooks/useHandleSseClientConnectionRetry.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useHandleSseClientConnectionRetry.ts new file mode 100644 index 0000000000..664d7337a3 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/useHandleSseClientConnectionRetry.ts @@ -0,0 +1,63 @@ +import { tokenPairState } from '@/auth/states/tokenPairState'; +import { SSE_CONNECTION_RETRY_MAX_WAIT_TIME_IN_MS } from '@/sse-db-event/constants/SseConnectionRetryMaxWaitTimeInMs'; +import { SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_FOR_DEV_MODE } from '@/sse-db-event/constants/SseConnectionRetryWaitTimeInMsForDevMode'; +import { SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_TO_AVOID_RACE_CONDITIONS } from '@/sse-db-event/constants/SseConnectionRetryWaitTimeInMsToAvoidRaceConditions'; +import { shouldDestroyEventStreamState } from '@/sse-db-event/states/shouldDestroyEventStreamState'; +import { sseClientState } from '@/sse-db-event/states/sseClientState'; +import { getSnapshotValue } from '@/ui/utilities/state/utils/getSnapshotValue'; +import { useRecoilCallback } from 'recoil'; +import { isDefined } from 'twenty-shared/utils'; +import { getIsDevelopmentEnvironment } from '~/utils/getIsDevelopmentEnvironment'; +import { sleep } from '~/utils/sleep'; + +export const useHandleSseClientConnectionRetry = () => { + const handleSseClientConnectionRetry = useRecoilCallback( + ({ snapshot, set }) => + async (retryCount: number, initialTokenForSseClient: string) => { + const sseClient = getSnapshotValue(snapshot, sseClientState); + + if (!isDefined(sseClient)) { + await sleep( + SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_TO_AVOID_RACE_CONDITIONS, + ); + + return; + } + + const tokenPair = getSnapshotValue(snapshot, tokenPairState); + const currentAppToken = + tokenPair?.accessOrWorkspaceAgnosticToken?.token; + + const shouldResetSseClient = + !isDefined(currentAppToken) || + currentAppToken !== initialTokenForSseClient || + retryCount > 10; + + if (shouldResetSseClient) { + await sleep( + SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_TO_AVOID_RACE_CONDITIONS, + ); + + sseClient.dispose(); + set(shouldDestroyEventStreamState, true); + set(sseClientState, null); + return; + } + + const randomWaitTimeInMsToSpaceAllClientsReconnection = Math.round( + Math.random() * SSE_CONNECTION_RETRY_MAX_WAIT_TIME_IN_MS, + ); + + const isDevelopmentEnvironment = getIsDevelopmentEnvironment(); + + const waitTimeInMs = isDevelopmentEnvironment + ? SSE_CONNECTION_RETRY_WAIT_TIME_IN_MS_FOR_DEV_MODE + : randomWaitTimeInMsToSpaceAllClientsReconnection; + + await sleep(waitTimeInMs); + }, + [], + ); + + return { handleSseClientConnectionRetry }; +}; 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 177ceee3e4..897b9a91da 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 @@ -5,12 +5,12 @@ import { disposeFunctionForEventStreamState } from '@/sse-db-event/states/dispos import { isCreatingSseEventStreamState } from '@/sse-db-event/states/isCreatingSseEventStreamState'; import { isDestroyingEventStreamState } from '@/sse-db-event/states/isDestroyingEventStreamState'; import { shouldDestroyEventStreamState } from '@/sse-db-event/states/shouldDestroyEventStreamState'; +import { sseClientState } from '@/sse-db-event/states/sseClientState'; import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdState'; import { getSnapshotValue } from '@/ui/utilities/state/utils/getSnapshotValue'; import { captureException } from '@sentry/react'; import { isNonEmptyString } from '@sniptt/guards'; import { print, type ExecutionResult } from 'graphql'; -import { type Client } from 'graphql-sse'; import { useRecoilCallback, useSetRecoilState } from 'recoil'; import { isDefined } from 'twenty-shared/utils'; @@ -30,7 +30,9 @@ export const useTriggerEventStreamCreation = () => { const triggerEventStreamCreation = useRecoilCallback( ({ snapshot, set }) => - (sseClient: Client) => { + () => { + const sseClient = snapshot.getLoadable(sseClientState).getValue(); + const isCreatingSseEventStream = snapshot .getLoadable(isCreatingSseEventStreamState) .getValue();