Merge pull request #240 from pneumaticapp/frontend/websocket/47593__Catch_errors_for_websockets

47593 frontend [ websocket ] Catch errors for websockets

Author: pneumatic-ilyacedrik

frontend/src/public/layout/MainLayout/MainLayout.tsx

@@ -27,9 +27,6 @@ import { isEnvAnalytics, isEnvPush } from '../../constants/enviroment';
 import { IUnsavedUser, TUserListItem } from '../../types/user';
 import { checkIsTemplateOwner, loadGroups } from '../../redux/actions';
 import { getIsAdmin } from '../../redux/selectors/user';
-import { closeAllConnections, hasActiveConnections } from '../../redux/utils/webSocketConnections';
-import { promiseDelay } from '../../utils/timeouts';
-
 import styles from './MainLayout.css';
 
 export interface IMainLayoutComponentStoreProps {
@@ -101,13 +98,7 @@ export function MainLayout({
 
   React.useEffect(() => {
     if (billingPlan) {
-      if (hasActiveConnections()) {
-        closeAllConnections()
-          .then(() => promiseDelay(500))
-          .then(() => watchUserWSEventsAction());
-      } else {
-        watchUserWSEventsAction();
-      }
+      watchUserWSEventsAction();
 
       loadNotificationsList();
       loadTasksCount();

frontend/src/public/redux/auth/saga.ts

@@ -1,4 +1,5 @@
-import { all, call, fork, put, takeEvery, takeLatest, select, takeLeading } from 'redux-saga/effects';
+import type { Task } from 'redux-saga';
+import { all, call, cancel, fork, put, takeEvery, takeLatest, select, takeLeading } from 'redux-saga/effects';
 import { auth } from '../../api/auth';
 import {
   accountEditFailed,
@@ -66,6 +67,7 @@ import { getUTMParams, IUserUtm } from '../../views/user/utils/utmParams';
 import { clearAppFilters, setGeneralLoaderVisibility } from '../general/actions';
 import { getQueryStringParams, history } from '../../utils/history';
 import { watchWsEvents } from '../realtime/watchWsEvents';
+import { closeAllConnections } from '../utils/webSocketConnections';
 import { TUploadedFile, uploadUserAvatar } from '../../utils/uploadFiles';
 import { changePhotoProfile } from '../../api/changePhotoProfile';
 import { ELoggedState, IAuthUser } from '../../types/redux';
@@ -268,6 +270,7 @@ export function* registerWithInvite({ payload }: TRegisterUserInvited) {
 }
 
 function* handleLogoutUnsubscribing({ shouldExpireToken }: { shouldExpireToken: boolean }) {
+  yield call(stopWatchWsEvents);
   yield put(clearAppFilters());
   resetFirebaseDeviceToken();
   resetSuperuserToken();
@@ -688,12 +691,24 @@ export function* watchSetUserToken() {
   yield takeEvery(EAuthActions.SetToken, handleSetUserToken);
 }
 
+let watchWsEventsTask: Task | undefined;
+
+function* stopWatchWsEvents() {
+  if (watchWsEventsTask) {
+    yield cancel(watchWsEventsTask);
+    watchWsEventsTask = undefined;
+  }
+
+  yield call(closeAllConnections);
+}
+
 function* handleWatchUserWSEvents() {
-  yield fork(watchWsEvents);
+  yield call(stopWatchWsEvents);
+  watchWsEventsTask = yield fork(watchWsEvents);
 }
 
 export function* watchUserWSEvents() {
-  yield takeEvery(EAuthActions.WatchUserWSEvents, handleWatchUserWSEvents);
+  yield takeLeading(EAuthActions.WatchUserWSEvents, handleWatchUserWSEvents);
 }
 
 export function* watchUploadUserPhoto() {

frontend/src/public/redux/realtime/utils/routeRealtimeEvent.ts

@@ -1,6 +1,7 @@
 import { call, put, select } from 'redux-saga/effects';
 
 import type { IRealtimeWsEnvelope } from '../types';
+import type { IStoreTask, IStoreWorkflows } from '../../../types/redux';
 import { ERealtimeEnvelopeType } from '../types';
 
 import { handleAddTask, handleRemoveTask } from '../../tasks/saga';
@@ -17,7 +18,6 @@ import { getUserTimezone } from '../../selectors/user';
 import { updateWorkflowLogItem } from '../../workflows/slice';
 import { isWorkflowEndedEventType } from './isWorkflowEndedEventType';
 import { mapBackendNewEventToRedux } from '../../../utils/mappers';
-import type { IStoreTask, IStoreWorkflows } from '../../../types/redux';
 import { getWorkflowsStore } from '../../selectors/workflows';
 import { getTaskStore } from '../../selectors/task';
 
@@ -86,16 +86,14 @@ export function* routeRealtimeEvent(envelope: IRealtimeWsEnvelope) {
       if (item) {
         yield call(prependNotificationItem, item);
       } else {
-        logger.error(`unhandled WebSocket event: ${envelope.type}, id=${envelope.id}`);
+        logger.error(`unhandled WebSocket event: ${envelope.type}`);
       }
 
       break;
     }
     default: {
       const unexpectedEnvelope = envelope as IRealtimeWsEnvelope;
-      logger.error(
-        `unhandled WebSocket event: ${unexpectedEnvelope.type}, id=${unexpectedEnvelope.id}`,
-      );
+      logger.error(`unhandled WebSocket event: ${String(unexpectedEnvelope.type)}`);
       break;
     }
   }

frontend/src/public/redux/realtime/watchWsEvents.ts

@@ -8,6 +8,7 @@ import { getBrowserConfigEnv } from '../../utils/getConfig';
 import { mergePaths } from '../../utils/urls';
 import type { IRealtimeWsEnvelope } from './types';
 import { routeRealtimeEvent } from './utils/routeRealtimeEvent';
+import { logger } from '../../utils/logger';
 
 
 
@@ -22,6 +23,10 @@ export function* watchWsEvents() {
 
   while (true) {
     const envelope: IRealtimeWsEnvelope = yield take(channel);
-    yield call(routeRealtimeEvent, envelope);
+    try {
+      yield call(routeRealtimeEvent, envelope);
+    } catch (error) {
+      logger.error(`failed to route WebSocket event`, error);
+    }
   }
 }

frontend/src/public/redux/utils/createWebSocketChannel.ts

@@ -1,5 +1,6 @@
 import { eventChannel } from 'redux-saga';
 import { mapToCamelCase } from '../../utils/mappers';
+import { logger } from '../../utils/logger';
 import { addConnection } from './webSocketConnections';
 
 export interface WebSocketWithRemoveFlag extends WebSocket {
@@ -54,8 +55,16 @@ export function createWebSocketChannel(url: string) {
           return;
         }
 
-        const data = mapToCamelCase(JSON.parse(message.data));
-        emit(data);
+        try {
+          const data = mapToCamelCase(JSON.parse(message.data));
+          emit(data);
+        } catch (error) {
+          logger.error('Invalid WebSocket message', message.data, error);
+        }
+      };
+
+      ws.onerror = (event) => {
+        logger.error('WebSocket connection error', event);
       };
 
       ws.onclose = () => {
@@ -72,6 +81,7 @@ export function createWebSocketChannel(url: string) {
         clearInterval(heartbeatInterval);
         heartbeatInterval = null;
       }
+      ws.shouldRemove = true;
       ws.close();
     };
   });