Return false instead of throwing when event stream does not exist (#20165)

## Summary

- When an event stream expires (TTL), `addQueryToEventStream` and
`removeQueryFromEventStream` now return `false` instead of throwing
`EVENT_STREAM_DOES_NOT_EXIST` as an `InternalServerError`
- Frontend checks the mutation return value and triggers the
destroy/recreate cycle, same recovery behavior without the error path
- Removes `EVENT_STREAM_DOES_NOT_EXIST` from exception code, exception
filter, and frontend graceful error check since it's no longer thrown

## Test plan

- [x] Verify that when an event stream TTL expires, the frontend
silently recreates the stream without error noise in logs/Sentry
- [x] Verify that `NOT_AUTHORIZED` errors still throw correctly on both
mutations
- [ ] Verify that the subscription `onEventSubscription` still works
end-to-end with stream creation, query registration, and heartbeat TTL
refresh

Made with [Cursor](https://cursor.com)

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

@@ -29,12 +29,12 @@ export const SSEQuerySubscribeEffect = () => {
   const sseEventStreamReady = useAtomStateValue(sseEventStreamReadyState);
 
   const [addQueryToEventStream] = useMutation<
-    boolean,
+    { addQueryToEventStream: boolean },
     { input: AddQuerySubscriptionInput }
   >(ADD_QUERY_TO_EVENT_STREAM_MUTATION);
 
   const [removeQueryFromEventStream] = useMutation<
-    void,
+    { removeQueryFromEventStream: boolean },
     { input: RemoveQueryFromEventStreamInput }
   >(REMOVE_QUERY_FROM_EVENT_STREAM_MUTATION);
 
@@ -66,7 +66,7 @@ export const SSEQuerySubscribeEffect = () => {
 
     try {
       for (const queryListenerToAdd of queryListenersToAdd) {
-        await addQueryToEventStream({
+        const result = await addQueryToEventStream({
           variables: {
             input: {
               eventStreamId: sseEventStreamId,
@@ -75,10 +75,17 @@ export const SSEQuerySubscribeEffect = () => {
             },
           },
         });
+
+        if (result.data?.addQueryToEventStream === false) {
+          store.set(activeQueryListenersState.atom, []);
+          store.set(shouldDestroyEventStreamState.atom, true);
+
+          return;
+        }
       }
 
       for (const queryListenerToRemove of queryListenersToRemove) {
-        await removeQueryFromEventStream({
+        const result = await removeQueryFromEventStream({
           variables: {
             input: {
               eventStreamId: sseEventStreamId,
@@ -86,6 +93,13 @@ export const SSEQuerySubscribeEffect = () => {
             },
           },
         });
+
+        if (result.data?.removeQueryFromEventStream === false) {
+          store.set(activeQueryListenersState.atom, []);
+          store.set(shouldDestroyEventStreamState.atom, true);
+
+          return;
+        }
       }
     } catch (error) {
       if (CombinedGraphQLErrors.is(error)) {
@@ -99,6 +113,7 @@ export const SSEQuerySubscribeEffect = () => {
         ) {
           store.set(activeQueryListenersState.atom, []);
           store.set(shouldDestroyEventStreamState.atom, true);
+
           return;
         }
 

packages/twenty-front/src/modules/sse-db-event/utils/isGracefullyHandledEventStreamError.ts

@@ -12,7 +12,6 @@ export const isGracefullyHandledEventStreamError = ({
   }
 
   return (
-    subCode === 'EVENT_STREAM_DOES_NOT_EXIST' ||
     subCode === 'EVENT_STREAM_ALREADY_EXISTS' ||
     subCode === 'NOT_AUTHORIZED' ||
     code === 'UNAUTHENTICATED' ||

packages/twenty-server/src/engine/subscriptions/event-stream-exception.filter.ts

@@ -14,7 +14,6 @@ export class EventStreamExceptionFilter implements GqlExceptionFilter {
   catch(exception: EventStreamException) {
     switch (exception.code) {
       case EventStreamExceptionCode.EVENT_STREAM_ALREADY_EXISTS:
-      case EventStreamExceptionCode.EVENT_STREAM_DOES_NOT_EXIST:
       case EventStreamExceptionCode.NOT_AUTHORIZED:
         throw new InternalServerError(exception.message, {
           subCode: exception.code,

packages/twenty-server/src/engine/subscriptions/event-stream.exception.ts

@@ -7,7 +7,6 @@ import { CustomException } from 'src/utils/custom-exception';
 export enum EventStreamExceptionCode {
   EVENT_STREAM_ALREADY_EXISTS = 'EVENT_STREAM_ALREADY_EXISTS',
   NOT_AUTHORIZED = 'NOT_AUTHORIZED',
-  EVENT_STREAM_DOES_NOT_EXIST = 'EVENT_STREAM_DOES_NOT_EXIST',
 }
 
 const getEventStreamExceptionUserFriendlyMessage = (
@@ -15,7 +14,6 @@ const getEventStreamExceptionUserFriendlyMessage = (
 ) => {
   switch (code) {
     case EventStreamExceptionCode.EVENT_STREAM_ALREADY_EXISTS:
-    case EventStreamExceptionCode.EVENT_STREAM_DOES_NOT_EXIST:
       return msg`Failed to receive real time updates.`;
     case EventStreamExceptionCode.NOT_AUTHORIZED:
       return msg`You are not authorized to perform this action.`;

packages/twenty-server/src/engine/subscriptions/event-stream.resolver.ts

@@ -151,11 +151,9 @@ export class EventStreamResolver {
     );
 
     if (!isDefined(streamData)) {
-      throw new EventStreamException(
-        'Event stream does not exist',
-        EventStreamExceptionCode.EVENT_STREAM_DOES_NOT_EXIST,
-      );
+      return false;
     }
+
     const isAuthorized = await this.eventStreamService.isAuthorized({
       streamData,
       authContext: {
@@ -170,6 +168,7 @@ export class EventStreamResolver {
         EventStreamExceptionCode.NOT_AUTHORIZED,
       );
     }
+
     await this.eventStreamService.addQuery({
       workspaceId: workspace.id,
       eventStreamChannelId,
@@ -197,10 +196,7 @@ export class EventStreamResolver {
     );
 
     if (!isDefined(streamData)) {
-      throw new EventStreamException(
-        'Event stream does not exist',
-        EventStreamExceptionCode.EVENT_STREAM_DOES_NOT_EXIST,
-      );
+      return false;
     }
 
     const isAuthorized = await this.eventStreamService.isAuthorized({