Fix event stream does not exists error (#17873)

[Event stream does not
exists](https://twenty-v7.sentry.io/issues/7238297246/events/441b3c0465cf475a8349c7dec40cae2a/?environment=prod&environment=prod-eu&project=4507072499810304&query=is%3Aunresolved%20%21issue.type%3A%5Bperformance_consecutive_db_queries%2Cperformance_consecutive_http%2Cperformance_file_io_main_thread%2Cperformance_db_main_thread%2Cperformance_n_plus_one_db_queries%2Cperformance_n_plus_one_api_calls%2Cperformance_p95_endpoint_regression%2Cperformance_slow_db_query%2Cperformance_render_blocking_asset_span%2Cperformance_uncompressed_assets%2Cperformance_http_overhead%2Cperformance_large_http_payload%5D%20timesSeen%3A%3E10&referrer=previous-event&sort=date)

Error happens when we are trying to add a query to a non-existing event
stream. In some cases, this is legit. Stream has expired and needs to be
re-created. Then we try to add the query again.

But it should happen only once per tab, and not often. We have a lot of
errors in sentry for each users.

Potential root cause: a race condition between the event stream creation
and the addition of queries:
- event stream id is created in frontend state + creation query is sent
- event stream id is in state so query can be added
- addQuery happens before the stream is actually created in redis. So an
error is returned
- the error makes the event stream re-generated by frontend 
- => the flow starts again until the event stream is actually created
BEFORE the first query is added

Fix: a new state saying if event stream is ready
- event stream id is created in frontend state BUT ready state is falsy
so query is not added yet
- on stream creation on backend side, an initial event is sent, so the
frontend knows the stream is ready
- addQuery can be triggered safely

packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx

@@ -4,6 +4,7 @@ import { activeQueryListenersState } from '@/sse-db-event/states/activeQueryList
 import { requiredQueryListenersState } from '@/sse-db-event/states/requiredQueryListenersState';
 import { shouldDestroyEventStreamState } from '@/sse-db-event/states/shouldDestroyEventStreamState';
 import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdState';
+import { sseEventStreamReadyState } from '@/sse-db-event/states/sseEventStreamReadyState';
 import { getSnapshotValue } from '@/ui/utilities/state/utils/getSnapshotValue';
 import { ApolloError, useMutation } from '@apollo/client';
 import { isNonEmptyString } from '@sniptt/guards';
@@ -21,6 +22,7 @@ import {
 
 export const SSEQuerySubscribeEffect = () => {
   const sseEventStreamId = useRecoilValue(sseEventStreamIdState);
+  const sseEventStreamReady = useRecoilValue(sseEventStreamReadyState);
 
   const [addQueryToEventStream] = useMutation<
     boolean,
@@ -122,7 +124,7 @@ export const SSEQuerySubscribeEffect = () => {
   );
 
   useEffect(() => {
-    if (!isNonEmptyString(sseEventStreamId)) {
+    if (!isNonEmptyString(sseEventStreamId) || !sseEventStreamReady) {
       return;
     }
 
@@ -138,6 +140,7 @@ export const SSEQuerySubscribeEffect = () => {
     }
   }, [
     sseEventStreamId,
+    sseEventStreamReady,
     requiredQueryListeners,
     activeQueryListeners,
     debouncedUpdateQueryListeners,

packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts

@@ -7,6 +7,7 @@ import { isDestroyingEventStreamState } from '@/sse-db-event/states/isDestroying
 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 { sseEventStreamReadyState } from '@/sse-db-event/states/sseEventStreamReadyState';
 import { getSnapshotValue } from '@/ui/utilities/state/utils/getSnapshotValue';
 import { captureException } from '@sentry/react';
 import { isNonEmptyString } from '@sniptt/guards';
@@ -60,6 +61,9 @@ export const useTriggerEventStreamCreation = () => {
         const newSseEventStreamId = v4();
 
         set(sseEventStreamIdState, newSseEventStreamId);
+        set(sseEventStreamReadyState, false);
+
+        let hasReceivedFirstEvent = false;
 
         const dispose = sseClient.subscribe(
           {
@@ -74,6 +78,11 @@ export const useTriggerEventStreamCreation = () => {
                 onEventSubscription: EventSubscription;
               }>,
             ) => {
+              if (!hasReceivedFirstEvent) {
+                hasReceivedFirstEvent = true;
+                set(sseEventStreamReadyState, true);
+              }
+
               const objectRecordEventsWithQueryIds =
                 value?.data?.onEventSubscription?.eventWithQueryIdsList ?? [];
 

packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamDestroy.ts

@@ -3,6 +3,7 @@ import { isCreatingSseEventStreamState } from '@/sse-db-event/states/isCreatingS
 import { isDestroyingEventStreamState } from '@/sse-db-event/states/isDestroyingEventStreamState';
 import { shouldDestroyEventStreamState } from '@/sse-db-event/states/shouldDestroyEventStreamState';
 import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdState';
+import { sseEventStreamReadyState } from '@/sse-db-event/states/sseEventStreamReadyState';
 import { isNonEmptyString } from '@sniptt/guards';
 import { useRecoilCallback, useSetRecoilState } from 'recoil';
 
@@ -40,6 +41,7 @@ export const useTriggerEventStreamDestroy = () => {
           disposeFunctionForEventStream?.dispose();
 
           set(sseEventStreamIdState, null);
+          set(sseEventStreamReadyState, false);
           set(disposeFunctionForEventStreamState, null);
           set(shouldDestroyEventStreamState, false);
         }

packages/twenty-front/src/modules/sse-db-event/states/sseEventStreamReadyState.ts

@@ -0,0 +1,6 @@
+import { createState } from 'twenty-ui/utilities';
+
+export const sseEventStreamReadyState = createState<boolean>({
+  key: 'sseEventStreamReadyState',
+  defaultValue: false,
+});

packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-lifecycle.ts

@@ -1,6 +1,7 @@
 import { isDefined } from 'twenty-shared/utils';
 
-type AsyncIteratorLifecycleOptions = {
+type AsyncIteratorLifecycleOptions<T> = {
+  initialValue?: T;
   onHeartbeat?: () => Promise<boolean>;
   heartbeatIntervalMs?: number;
   onCleanup?: () => Promise<void>;
@@ -8,10 +9,11 @@ type AsyncIteratorLifecycleOptions = {
 
 export function wrapAsyncIteratorWithLifecycle<T>(
   iterator: AsyncIterableIterator<T>,
-  options: AsyncIteratorLifecycleOptions,
+  options: AsyncIteratorLifecycleOptions<T>,
 ): AsyncIterableIterator<T> {
-  const { onHeartbeat, heartbeatIntervalMs, onCleanup } = options;
+  const { initialValue, onHeartbeat, heartbeatIntervalMs, onCleanup } = options;
   let heartbeatInterval: NodeJS.Timeout | null = null;
+  let hasYieldedInitialValue = false;
 
   const startHeartbeat = () => {
     if (onHeartbeat && heartbeatIntervalMs) {
@@ -41,6 +43,12 @@ export function wrapAsyncIteratorWithLifecycle<T>(
         startHeartbeat();
       }
 
+      if (isDefined(initialValue) && !hasYieldedInitialValue) {
+        hasYieldedInitialValue = true;
+
+        return { done: false, value: initialValue };
+      }
+
       let result: IteratorResult<T>;
 
       try {

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts

@@ -142,6 +142,7 @@ export class WorkspaceEventEmitterResolver {
     }
 
     return wrapAsyncIteratorWithLifecycle(iterator, {
+      initialValue: [],
       onHeartbeat: () =>
         this.eventStreamService.refreshEventStreamTTL({
           workspaceId: workspace.id,