feat: Optimized Socket Connection and Too many socket connection prev… (#1003)

…ention

Author: chavda-bhavik

apps/api/src/app/shared/services/websocket-service/websocket.service.ts

@@ -110,11 +110,6 @@ export class WebSocketService implements OnGatewayConnection, OnGatewayDisconnec
     });
   }
 
-  // Get abort signal for a session
-  getSessionAbortSignal(sessionId: string): AbortSignal | undefined {
-    return this.sessionAbortControllers.get(sessionId)?.signal;
-  }
-
   // Use arrow functions to automatically bind 'this'
   sendProgress = (sessionId: string, progressData: IProgressData) => {
     if (!this.server) {
@@ -145,7 +140,7 @@ export class WebSocketService implements OnGatewayConnection, OnGatewayDisconnec
         error,
         timestamp: new Date().toISOString(),
       });
-      this.logger.error(`✅ Error sent to session ${sessionId}: ${error}`);
+      this.logger.error(`❌ Error sent to session ${sessionId}: ${error}`);
     } catch (err) {
       this.logger.error(`Failed to send error to session ${sessionId}:`, err);
     } finally {
@@ -172,38 +167,4 @@ export class WebSocketService implements OnGatewayConnection, OnGatewayDisconnec
       this.sessionAbortControllers.delete(sessionId);
     }
   };
-
-  // Helper method to check if a session has connected clients
-  hasClientsInSession = (sessionId: string): boolean => {
-    const room = this.server.sockets.adapter.rooms.get(sessionId);
-
-    return room && room.size > 0;
-  };
-
-  // Method to get session info for debugging
-  getSessionInfo = (sessionId: string) => {
-    const room = this.server.sockets.adapter.rooms.get(sessionId);
-    const hasAbortController = this.sessionAbortControllers.has(sessionId);
-    const isAborted = this.sessionAbortControllers.get(sessionId)?.signal.aborted;
-
-    return {
-      sessionId,
-      clientCount: room ? room.size : 0,
-      hasClients: room && room.size > 0,
-      hasAbortController,
-      isAborted,
-    };
-  };
-
-  // Method to manually abort all sessions (useful for cleanup)
-  abortAllSessions() {
-    this.logger.log('Aborting all active sessions');
-    this.sessionAbortControllers.forEach((controller, sessionId) => {
-      if (!controller.signal.aborted) {
-        controller.abort();
-        this.logger.log(`Aborted session ${sessionId}`);
-      }
-    });
-    this.sessionAbortControllers.clear();
-  }
 }

apps/widget/src/components/widget/Phases/AutoImport/AutoImportPhase1/AutoImportPhase1.tsx

@@ -1,51 +1,35 @@
-/* eslint-disable @typescript-eslint/no-unused-vars */
-/* eslint-disable no-unused-vars */
 import { Stack, TextInput, Text } from '@mantine/core';
-import { PhasesEnum } from '@types';
 import { validateRssUrl } from '@util';
+import { PhasesEnum } from '@types';
 import { Footer } from 'components/Common/Footer';
 import { useAutoImportPhase1 } from '@hooks/AutoImportPhase1/useAutoImportPhase1';
 import { ProgressRing } from './ProgressRing';
-import { ConfirmModal } from 'components/widget/modals/ConfirmModal';
-import { WIDGET_TEXTS } from '@impler/client';
-import { useState, useCallback, useEffect } from 'react';
+import { colors } from '@config';
 
 interface IAutoImportPhase1Props {
   onNextClick: () => void;
   onCloseClick: () => void;
   onRssParsingStart?: () => void;
   onRssParsingEnd?: () => void;
   onRegisterAbortFunction?: (abortFn: () => void) => void;
+  onRegisterDisconnectFunction?: (disconnectFn: () => void) => void;
 }
 
 export function AutoImportPhase1({
   onNextClick,
   onRssParsingStart,
   onRssParsingEnd,
   onRegisterAbortFunction,
+  onRegisterDisconnectFunction,
 }: IAutoImportPhase1Props) {
-  const { isGetRssXmlHeadingsLoading, progressPercentage, register, errors, onSubmit, abortOperation, canAbort } =
-    useAutoImportPhase1({
-      goNext: onNextClick,
-    });
-
-  // Register the abort function with the parent component
-  useEffect(() => {
-    if (canAbort && abortOperation && onRegisterAbortFunction) {
-      onRegisterAbortFunction(abortOperation);
-    }
-  }, [canAbort, abortOperation, onRegisterAbortFunction]);
-
-  // Notify parent component about RSS parsing status
-  useEffect(() => {
-    if (isGetRssXmlHeadingsLoading) {
-      onRssParsingStart?.();
-    } else {
-      onRssParsingEnd?.();
-    }
-  }, [isGetRssXmlHeadingsLoading, onRssParsingStart, onRssParsingEnd]);
+  const { isGetRssXmlHeadingsLoading, progressPercentage, register, errors, onSubmit, texts } = useAutoImportPhase1({
+    goNext: onNextClick,
+    onRegisterAbortFunction,
+    onRegisterDisconnectFunction,
+    onRssParsingStart,
+    onRssParsingEnd,
+  });
 
-  // Show form view when not processing
   return (
     <Stack spacing="xs" style={{ height: '100%', justifyContent: 'space-between' }}>
       <form onSubmit={onSubmit}>
@@ -70,6 +54,9 @@ export function AutoImportPhase1({
       {isGetRssXmlHeadingsLoading && (
         <Stack align="center" justify="center" spacing="xl">
           <ProgressRing percentage={progressPercentage} />
+          <Text fw="bolder" color={colors.StrokeLight}>
+            {texts.AUTOIMPORT_PHASE1.CLOSE_CONFIRMATION.TITLE}
+          </Text>
         </Stack>
       )}
 

apps/widget/src/components/widget/Widget.tsx

@@ -37,6 +37,7 @@ export function Widget() {
 
   // Add ref to store abort function from AutoImportPhase1
   const abortRssOperationRef = useRef<(() => void) | null>(null);
+  const disconnectSocketRef = useRef<(() => void) | null>(null);
 
   const {
     flow,
@@ -63,25 +64,40 @@ export function Widget() {
       abortRssOperationRef.current();
     }
 
+    if (promptContinueAction === PromptModalTypesEnum.CLOSE && disconnectSocketRef.current) {
+      disconnectSocketRef.current();
+    }
+
     terminateUpload();
+
     setPromptContinueAction(undefined);
     if (uploadInfo._id) ParentWindow.UploadTerminated({ uploadId: uploadInfo._id });
     if (promptContinueAction === PromptModalTypesEnum.CLOSE) closeWidget();
   };
 
+  // Add callback to register disconnect function from AutoImportPhase1
+  const handleRegisterDisconnectFunction = useCallback((disconnectFn: () => void) => {
+    if (promptContinueAction === PromptModalTypesEnum.CLOSE) {
+      console.log('disconnectFn');
+    }
+    disconnectSocketRef.current = disconnectFn;
+  }, []);
+
   const onPromptCancel = () => {
     setPromptContinueAction(undefined);
   };
 
   const onClose = () => {
     let isImportNotOnProgress = false;
 
+    // If RSS is actively parsing, show confirmation dialog but DON'T disconnect socket yet
     if (flow === FlowsEnum.AUTO_IMPORT && phase === PhasesEnum.CONFIGURE && isRssParsing) {
       setPromptContinueAction(PromptModalTypesEnum.CLOSE);
 
       return;
     }
 
+    // For other AUTO_IMPORT phases
     if (flow === FlowsEnum.AUTO_IMPORT)
       isImportNotOnProgress = [PhasesEnum.CONFIGURE, PhasesEnum.CONFIRM].includes(phase);
     else if (flow == FlowsEnum.MANUAL_ENTRY)
@@ -95,10 +111,13 @@ export function Widget() {
       ].includes(phase);
 
     if (isImportNotOnProgress) {
+      if (disconnectSocketRef.current) disconnectSocketRef.current();
       setPhase(PhasesEnum.VALIDATE);
       resetAppState();
       closeWidget();
-    } else setPromptContinueAction(PromptModalTypesEnum.CLOSE);
+    } else {
+      setPromptContinueAction(PromptModalTypesEnum.CLOSE);
+    }
   };
 
   const closeWidget = () => {
@@ -108,6 +127,8 @@ export function Widget() {
     setIsRssParsing(false);
     // Clear the abort function reference
     abortRssOperationRef.current = null;
+    // Clear the disconnect function reference
+    disconnectSocketRef.current = null;
     setTimeout(() => {
       ParentWindow.Close();
     }, variables.closeDelayInMS);
@@ -119,6 +140,7 @@ export function Widget() {
     setPhase(PhasesEnum.VALIDATE);
     setIsRssParsing(false); // Reset RSS parsing state
     abortRssOperationRef.current = null; // Clear abort function reference
+    disconnectSocketRef.current = null; // Clear disconnect function reference
   };
 
   const onImportJobCreated = (jobInfo: IUserJob) => {
@@ -164,6 +186,7 @@ export function Widget() {
               onRssParsingStart={() => setIsRssParsing(true)}
               onRssParsingEnd={() => setIsRssParsing(false)}
               onRegisterAbortFunction={handleRegisterAbortFunction}
+              onRegisterDisconnectFunction={handleRegisterDisconnectFunction}
             />
           ),
           [PhasesEnum.MAPCOLUMNS]: <AutoImportPhase2 texts={texts} onNextClick={() => setPhase(PhasesEnum.SCHEDULE)} />,
@@ -210,6 +233,7 @@ export function Widget() {
       setPhase(PhasesEnum.VALIDATE);
       setIsRssParsing(false); // Reset RSS parsing state when widget is hidden
       abortRssOperationRef.current = null; // Clear abort function reference
+      disconnectSocketRef.current = null; // Clear disconnect function reference
     }
   }, [showWidget]);
 

apps/widget/src/hooks/AutoImportPhase1/useAutoImportPhase1.tsx

@@ -1,6 +1,6 @@
-import { useMutation } from '@tanstack/react-query';
-import { useForm, SubmitHandler } from 'react-hook-form';
 import { useCallback, useEffect, useMemo, useRef } from 'react';
+import { useForm, SubmitHandler } from 'react-hook-form';
+import { useMutation } from '@tanstack/react-query';
 
 import { generateSessionId, notifier } from '@util';
 import { IAutoImportValues, IErrorData } from '@types';
@@ -13,16 +13,26 @@ import { useWebSocketProgress } from '@hooks/AutoImportPhase1/useWebSocketProgre
 
 interface IUseAutoImportPhase1Props {
   goNext: () => void;
+  onRegisterAbortFunction?: (abortFn: () => void) => void;
+  onRegisterDisconnectFunction?: (disconnectFn: () => void) => void;
+  onRssParsingStart?: () => void;
+  onRssParsingEnd?: () => void;
 }
 
 interface FormValues {
   rssUrl: string;
 }
 
-export function useAutoImportPhase1({ goNext }: IUseAutoImportPhase1Props) {
+export function useAutoImportPhase1({
+  goNext,
+  onRegisterAbortFunction,
+  onRegisterDisconnectFunction,
+  onRssParsingStart,
+  onRssParsingEnd,
+}: IUseAutoImportPhase1Props) {
   const { api } = useAPIState();
   const { setJobsInfo } = useJobsInfo();
-  const { output, schema } = useAppState();
+  const { output, schema, texts } = useAppState();
   const { templateId, extra, authHeaderValue } = useImplerState();
 
   const webSocketSessionIdRef = useRef<string | null>(null);
@@ -62,13 +72,22 @@ export function useAutoImportPhase1({ goNext }: IUseAutoImportPhase1Props) {
   }, []);
 
   // Initialize WebSocket connection
-  const { isConnected, progressData, completionData, errorData, joinSession, leaveSession, socket, clearProgress } =
-    useWebSocketProgress({
-      onCompletion: handleCompletion,
-      onError: handleError,
-      onConnectionChange: handleConnectionChange,
-      autoConnect: true,
-    });
+  const {
+    isConnected,
+    progressData,
+    completionData,
+    errorData,
+    joinSession,
+    leaveSession,
+    socket,
+    disconnect,
+    clearProgress,
+  } = useWebSocketProgress({
+    onCompletion: handleCompletion,
+    onError: handleError,
+    onConnectionChange: handleConnectionChange,
+    autoConnect: true,
+  });
 
   // Mutation for RSS XML processing
   const { isLoading: isGetRssXmlHeadingsLoading, mutate: getRssXmlHeading } = useMutation<
@@ -189,21 +208,24 @@ export function useAutoImportPhase1({ goNext }: IUseAutoImportPhase1Props) {
     return 0;
   }, [progressData?.percentage]);
 
-  const handleCleanup = useCallback(() => {
-    // First abort the session if it's running
-    if (canAbort && webSocketSessionIdRef.current) {
-      abortOperation();
+  useEffect(() => {
+    if (canAbort && abortOperation && onRegisterAbortFunction) {
+      onRegisterAbortFunction(abortOperation);
     }
+  }, [canAbort, abortOperation, onRegisterAbortFunction]);
 
-    // Then do additional cleanup
-    if (socket && socket.connected) {
-      socket.disconnect();
+  // Register the disconnect function with the parent component
+  useEffect(() => {
+    if (isGetRssXmlHeadingsLoading) {
+      onRssParsingStart?.();
+    } else {
+      onRssParsingEnd?.();
     }
-    if (webSocketSessionIdRef.current) {
-      leaveSession(webSocketSessionIdRef.current);
-      webSocketSessionIdRef.current = null;
+
+    if (disconnect && onRegisterDisconnectFunction) {
+      onRegisterDisconnectFunction(disconnect);
     }
-  }, [socket, webSocketSessionIdRef, leaveSession, abortOperation, canAbort]);
+  }, [isGetRssXmlHeadingsLoading, onRssParsingStart, onRssParsingEnd, disconnect, onRegisterDisconnectFunction]);
 
   useEffect(() => {
     return () => {
@@ -225,6 +247,7 @@ export function useAutoImportPhase1({ goNext }: IUseAutoImportPhase1Props) {
 
     // WebSocket connection state
     isConnected,
+    disconnect,
     socket,
     leaveSession,
     webSocketSessionIdRef,
@@ -240,7 +263,8 @@ export function useAutoImportPhase1({ goNext }: IUseAutoImportPhase1Props) {
     // Abort functionality
     abortOperation,
     canAbort,
-    handleCleanup,
     isAborted: isAbortedRef.current,
+
+    texts,
   };
 }