Why am i getting disconnected from websockets when a regular user tries to subscribe to a topic with a different userid than the regular user has? It should show an error message but not disconnect. Can you please help me? This is what the backend terminal says:

2024-12-02T17:02:06.125+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] c.f.ecommerce.backend.Application        : Starting Application using Java 17.0.11 with PID 25252 (C:\Users\hassa\Documents\ecommerce.backend\target\classes started by hassa in C:\Users\hassa\Documents\ecommerce.backend)
2024-12-02T17:02:06.129+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] c.f.ecommerce.backend.Application        : Running with Spring Boot v3.3.5, Spring v6.1.14
2024-12-02T17:02:06.129+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] c.f.ecommerce.backend.Application        : No active profile set, falling back to 1 default profile: "default"
2024-12-02T17:02:06.169+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] .e.DevToolsPropertyDefaultsPostProcessor : Devtools property defaults active! Set 'spring.devtools.add-properties' to 'false' to disable
2024-12-02T17:02:06.169+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] .e.DevToolsPropertyDefaultsPostProcessor : For additional web related logging consider setting the 'logging.level.web' property to 'DEBUG'
2024-12-02T17:02:06.997+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data JPA repositories in DEFAULT mode.
2024-12-02T17:02:07.071+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 64 ms. Found 7 JPA repository interfaces.    
2024-12-02T17:02:08.042+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port 8080 (http)
2024-12-02T17:02:08.059+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
2024-12-02T17:02:08.059+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.apache.catalina.core.StandardEngine    : Starting Servlet engine: [Apache Tomcat/10.1.31]
2024-12-02T17:02:08.127+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
2024-12-02T17:02:08.127+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 1957 ms
2024-12-02T17:02:08.661+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.hibernate.jpa.internal.util.LogHelper  : HHH000204: Processing PersistenceUnitInfo [name: default]
2024-12-02T17:02:08.781+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] org.hibernate.Version                    : HHH000412: Hibernate ORM core version 6.5.3.Final
2024-12-02T17:02:08.840+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.h.c.internal.RegionFactoryInitiator    : HHH000026: Second-level cache disabled
2024-12-02T17:02:09.148+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.o.j.p.SpringPersistenceUnitInfo      : No LoadTimeWeaver setup: ignoring JPA class transformer
2024-12-02T17:02:09.174+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Starting...
2024-12-02T17:02:09.397+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] com.zaxxer.hikari.pool.HikariPool        : HikariPool-1 - Added connection ConnectionID:1 ClientConnectionId: 93133e9f-1d72-48a6-a659-a20657b39d85
2024-12-02T17:02:09.399+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Start completed.    
2024-12-02T17:02:09.423+01:00  WARN 25252 --- [ecommerce.backend] [  restartedMain] org.hibernate.orm.deprecation            : HHH90000025: SQLServerDialect does not need to be specified explicitly using 'hibernate.dialect' (remove the property setting and it will be selected by default)
2024-12-02T17:02:10.390+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.h.e.t.j.p.i.JtaPlatformInitiator       : HHH000489: No JTA platform available (set 'hibernate.transaction.jta.platform' to enable JTA platform integration)
2024-12-02T17:02:10.945+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] j.LocalContainerEntityManagerFactoryBean : Initialized JPA EntityManagerFactory for persistence unit 'default'
2024-12-02T17:02:11.414+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] c.f.e.b.api.security.JWTRequestFilter    : Filter 'JWTRequestFilter' configured for use
2024-12-02T17:02:11.966+01:00  WARN 25252 --- [ecommerce.backend] [  restartedMain] JpaBaseConfiguration$JpaWebConfiguration : spring.jpa.open-in-view is enabled by default. Therefore, database queries may be performed during view rendering. Explicitly configure spring.jpa.open-in-view to disable this warning
2024-12-02T17:02:12.027+01:00  WARN 25252 --- [ecommerce.backend] [  restartedMain] .s.s.UserDetailsServiceAutoConfiguration :

Using generated security password: e5b6a2eb-01e7-446b-8889-c26e5bcf45cf

This generated password is for development use only. Your security configuration must be updated before running your application in production.

2024-12-02T17:02:12.053+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] r$InitializeUserDetailsManagerConfigurer : Global AuthenticationManager configured with UserDetailsService bean with name inMemoryUserDetailsManager
2024-12-02T17:02:12.152+01:00 TRACE 25252 --- [ecommerce.backend] [  restartedMain] o.s.w.s.s.s.WebSocketHandlerMapping      : No patterns in 'webSocketHandlerMapping'
2024-12-02T17:02:12.235+01:00 TRACE 25252 --- [ecommerce.backend] [  restartedMain] o.s.w.s.s.s.WebSocketHandlerMapping      : Mapped [/websocket/**] onto org.springframework.web.socket.sockjs.support.SockJsHttpRequestHandler@502d56cd
2024-12-02T17:02:12.235+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] o.s.w.s.s.s.WebSocketHandlerMapping      : Patterns [/websocket/**] in 'stompWebSocketHandlerMapping'
2024-12-02T17:02:12.558+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 1 endpoint beneath base path '/actuator'
2024-12-02T17:02:12.681+01:00 TRACE 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler : Looking for @MessageExceptionHandler mappings: org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@61be1e9c, started on Mon Dec 02 17:02:06 CET 2024     
2024-12-02T17:02:12.687+01:00 TRACE 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler : Detected @MessageExceptionHandler methods in org.springframework.web.socket.messaging.WebSocketAnnotationMethodMessageHandler$MessagingControllerAdviceBean@2f04c461
2024-12-02T17:02:12.698+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.c.a.AuthenticationController:       

2024-12-02T17:02:12.699+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.c.CartController:

2024-12-02T17:02:12.699+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.c.ImageController:

2024-12-02T17:02:12.700+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.c.o.OrderController:

2024-12-02T17:02:12.700+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.c.p.ProductController:

2024-12-02T17:02:12.701+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.c.u.UserController:

2024-12-02T17:02:12.705+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.c.WebsocketController:
        {[MESSAGE],[/sendMessage]}: handleMessage(ChatMessage,String,StompHeaderAccessor)
        {[MESSAGE],[/test]}: handleTest(String)       
        {[MESSAGE],[/admin/message]}: handleAdminMessage(ChatMessage,String,StompHeaderAccessor)
2024-12-02T17:02:12.706+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        c.f.e.b.a.s.WebSocketController:
        {[MESSAGE],[/profile]}: handleProfile(String) 
2024-12-02T17:02:12.707+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] .WebSocketAnnotationMethodMessageHandler :
        o.s.b.a.w.s.e.BasicErrorController:

2024-12-02T17:02:13.105+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.b.d.a.OptionalLiveReloadServer       : LiveReload server is running on port 35729
2024-12-02T17:02:13.161+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port 8080 (http) with context path '/'
2024-12-02T17:02:13.162+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.ExecutorSubscribableChannel      : clientOutboundChannel added SubProtocolWebSocketHandler[StompSubProtocolHandler[v10.stomp, v11.stomp, v12.stomp]]
2024-12-02T17:02:13.163+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.ExecutorSubscribableChannel      : clientInboundChannel added WebSocketAnnotationMethodMessageHandler[prefixes=[/app/]]     
2024-12-02T17:02:13.163+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.b.SimpleBrokerMessageHandler     : Starting...
2024-12-02T17:02:13.163+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.ExecutorSubscribableChannel      : clientInboundChannel added SimpleBrokerMessageHandler [org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry@1fb56b6a]
2024-12-02T17:02:13.164+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.ExecutorSubscribableChannel      : brokerChannel added SimpleBrokerMessageHandler [org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry@1fb56b6a]
2024-12-02T17:02:13.164+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.b.SimpleBrokerMessageHandler     : BrokerAvailabilityEvent[available=true, SimpleBrokerMessageHandler [org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry@1fb56b6a]]
2024-12-02T17:02:13.164+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.b.SimpleBrokerMessageHandler     : Started.
2024-12-02T17:02:13.165+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.ExecutorSubscribableChannel      : clientInboundChannel added UserDestinationMessageHandler[DefaultUserDestinationResolver[prefix=/user/]]
2024-12-02T17:02:13.166+01:00 DEBUG 25252 --- [ecommerce.backend] [  restartedMain] o.s.m.s.ExecutorSubscribableChannel      : brokerChannel added UserDestinationMessageHandler[DefaultUserDestinationResolver[prefix=/user/]]
2024-12-02T17:02:13.175+01:00  INFO 25252 --- [ecommerce.backend] [  restartedMain] c.f.ecommerce.backend.Application        : Started Application in 7.327 seconds (process running for 7.746)
2024-12-02T17:02:15.104+01:00  INFO 25252 --- [ecommerce.backend] [on(2)-127.0.0.1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
2024-12-02T17:02:15.104+01:00  INFO 25252 --- [ecommerce.backend] [on(2)-127.0.0.1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
2024-12-02T17:02:15.106+01:00  INFO 25252 --- [ecommerce.backend] [on(2)-127.0.0.1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 1 ms   
2024-12-02T17:02:20.833+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-1] o.s.w.s.s.s.WebSocketHandlerMapping      : Mapped to HandlerExecutionChain with [org.springframework.web.socket.sockjs.support.SockJsHttpRequestHandler@502d56cd] and 1 interceptors      
2024-12-02T17:02:20.838+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-1] o.s.w.s.s.t.h.DefaultSockJsService       : Processing transport request: GET http://localhost:8080/websocket/websocket
2024-12-02T17:02:20.840+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-1] o.s.w.s.s.s.DefaultHandshakeHandler      : Processing request http://localhost:8080/websocket/websocket with headers=[sec-websocket-version:"13", sec-websocket-key:"pcpDsAj4d2YDjwtcIzfwwQ==", connection:"Upgrade", upgrade:"websocket", authorization:"Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJyb2xlIjoiVVNFUiIsImlzcyI6ImVDb21tZXJjZSIsIlVTRVJOQU1FIjoiaGVsbG8iLCJleHAiOjE3MzM3NDE4NDV9.-QNt0_x-g-RNH8cZ8sqUMrritahS-yZlJyMyjrO81dk", destination:"/app/sendMessage", sec-websocket-extensions:"permessage-deflate; client_max_window_bits", host:"localhost:8080", Content-Type:"application/json;charset=UTF-8"]
2024-12-02T17:02:20.841+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-1] o.s.w.s.s.s.DefaultHandshakeHandler      : Upgrading to WebSocket, subProtocol=null, extensions=[]
2024-12-02T17:02:20.864+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-1] s.w.s.h.LoggingWebSocketHandlerDecorator : New StandardWebSocketSession[id=59e1bbd2-7469-7e0c-772e-be7aef87d467, uri=ws://localhost:8080/websocket/websocket]
2024-12-02T17:02:20.869+01:00  INFO 25252 --- [ecommerce.backend] [nio-8080-exec-1] c.f.e.b.a.s.WebSocketSessionRegistry     : Session registered for user: hello 
2024-12-02T17:02:21.783+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-2] s.w.s.h.LoggingWebSocketHandlerDecorator : Handling TextMessage payload=[CONNECT
a..], byteCount=251, last=true] in StandardWebSocketSession[id=59e1bbd2-7469-7e0c-772e-be7aef87d467, uri=ws://localhost:8080/websocket/websocket]
2024-12-02T17:02:21.787+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-2] o.s.messaging.simp.stomp.StompDecoder    : Decoded CONNECT {accept-version=[1.1,1.0], heart-beat=[10000,10000], Authorization=[Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJyb2xlIjoiVVNFUiIsImlzcyI6ImVDb21tZXJjZSIsIlVTRVJOQU1FIjoiaGVsbG8iLCJleHAiOjE3MzM3NDE4NDV9.-QNt0_x-g-RNH8cZ8sqUMrritahS-yZlJyMyjrO81dk]} session=null
2024-12-02T17:02:21.789+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-2] o.s.w.s.m.StompSubProtocolHandler        : From client: CONNECT user=hello session=59e1bbd2-7469-7e0c-772e-be7aef87d467
2024-12-02T17:02:21.794+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-2] c.f.e.b.a.s.WebSocketChannelInterceptor  : Processing CONNECT command
2024-12-02T17:02:21.794+01:00  INFO 25252 --- [ecommerce.backend] [nio-8080-exec-2] c.f.e.b.a.s.WebSocketChannelInterceptor  : User hello connected
2024-12-02T17:02:21.795+01:00 DEBUG 25252 --- [ecommerce.backend] [nboundChannel-2] o.s.m.s.b.SimpleBrokerMessageHandler     : Processing CONNECT user=hello session=59e1bbd2-7469-7e0c-772e-be7aef87d467
2024-12-02T17:02:21.799+01:00 TRACE 25252 --- [ecommerce.backend] [tboundChannel-1] o.s.messaging.simp.stomp.StompEncoder    : Encoding STOMP CONNECTED, headers={version=[1.1], heart-beat=[25000,25000], user-name=[hello]}
2024-12-02T17:02:21.799+01:00 TRACE 25252 --- [ecommerce.backend] [tboundChannel-1] o.s.w.s.adapter.NativeWebSocketSession   : Sending TextMessage payload=[CONNECTED
..], byteCount=63, last=true], StandardWebSocketSession[id=59e1bbd2-7469-7e0c-772e-be7aef87d467, uri=ws://localhost:8080/websocket/websocket]
2024-12-02T17:02:24.604+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-3] s.w.s.h.LoggingWebSocketHandlerDecorator : Handling TextMessage payload=[SUBSC..], byteCount=55, last=true] in StandardWebSocketSession[id=59e1bbd2-7469-7e0c-772e-be7aef87d467, uri=ws://localhost:8080/websocket/websocket]
2024-12-02T17:02:24.605+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-3] o.s.messaging.simp.stomp.StompDecoder    : Decoded SUBSCRIBE {id=[sub-0], destination=[/topic/user/13/address]} session=null        
2024-12-02T17:02:24.605+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-3] o.s.w.s.m.StompSubProtocolHandler        : From client: SUBSCRIBE /topic/user/13/address id=sub-0 session=59e1bbd2-7469-7e0c-772e-be7aef87d467
2024-12-02T17:02:24.608+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-3] c.f.e.b.a.s.WebSocketSecurityService     : User hello access to channel 13: false
2024-12-02T17:02:24.610+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-3] o.s.w.s.m.StompSubProtocolHandler        : Failed to send message to MessageChannel in session 59e1bbd2-7469-7e0c-772e-be7aef87d467 

org.springframework.messaging.MessageDeliveryException: Failed to send message to ExecutorSubscribableChannel[clientInboundChannel]
        at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:149) ~[spring-messaging-6.1.14.jar:6.1.14]
        at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:125) ~[spring-messaging-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.messaging.StompSubProtocolHandler.handleMessageFromClient(StompSubProtocolHandler.java:343) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.messaging.SubProtocolWebSocketHandler.handleMessage(SubProtocolWebSocketHandler.java:356) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.handler.WebSocketHandlerDecorator.handleMessage(WebSocketHandlerDecorator.java:75) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.handler.WebSocketHandlerDecorator.handleMessage(WebSocketHandlerDecorator.java:75) ~[spring-websocket-6.1.14.jar:6.1.14]
        at com.foodstore.ecommerce.backend.api.security.EmaWebSocketHandlerDecorator.handleMessage(EmaWebSocketHandlerDecorator.java:27) ~[classes/:na]
        at org.springframework.web.socket.handler.WebSocketHandlerDecorator.handleMessage(WebSocketHandlerDecorator.java:75) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.handler.LoggingWebSocketHandlerDecorator.handleMessage(LoggingWebSocketHandlerDecorator.java:56) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.handler.ExceptionWebSocketHandlerDecorator.handleMessage(ExceptionWebSocketHandlerDecorator.java:58) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.adapter.standard.StandardWebSocketHandlerAdapter.handleTextMessage(StandardWebSocketHandlerAdapter.java:113) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.adapter.standard.StandardWebSocketHandlerAdapter$3.onMessage(StandardWebSocketHandlerAdapter.java:84) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.springframework.web.socket.adapter.standard.StandardWebSocketHandlerAdapter$3.onMessage(StandardWebSocketHandlerAdapter.java:81) ~[spring-websocket-6.1.14.jar:6.1.14]
        at org.apache.tomcat.websocket.WsFrameBase.sendMessageText(WsFrameBase.java:390) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.server.WsFrameServer.sendMessageText(WsFrameServer.java:130) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.WsFrameBase.processDataText(WsFrameBase.java:484) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.WsFrameBase.processData(WsFrameBase.java:284) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.WsFrameBase.processInputBuffer(WsFrameBase.java:130) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.server.WsFrameServer.onDataAvailable(WsFrameServer.java:85) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.server.WsFrameServer.doOnDataAvailable(WsFrameServer.java:184) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.server.WsFrameServer.notifyDataAvailable(WsFrameServer.java:164) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]
        at org.apache.tomcat.websocket.server.WsHttpUpgradeHandler.upgradeDispatch(WsHttpUpgradeHandler.java:152) ~[tomcat-embed-websocket-10.1.31.jar:10.1.31]   
        at org.apache.coyote.http11.upgrade.UpgradeProcessorInternal.dispatch(UpgradeProcessorInternal.java:60) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:57) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:905) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1741) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:52) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at org.apache.tomcat.util.threads.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1190) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at org.apache.tomcat.util.threads.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:659) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:63) ~[tomcat-embed-core-10.1.31.jar:10.1.31]
        at java.base/java.lang.Thread.run(Thread.java:842) ~[na:na]
Caused by: org.springframework.security.access.AccessDeniedException: Access is denied
        at org.springframework.security.access.vote.AffirmativeBased.decide(AffirmativeBased.java:77) ~[spring-security-core-6.3.4.jar:6.3.4]
        at org.springframework.security.access.intercept.AbstractSecurityInterceptor.attemptAuthorization(AbstractSecurityInterceptor.java:253) ~[spring-security-core-6.3.4.jar:6.3.4]
        at org.springframework.security.access.intercept.AbstractSecurityInterceptor.beforeInvocation(AbstractSecurityInterceptor.java:222) ~[spring-security-core-6.3.4.jar:6.3.4]
        at org.springframework.security.messaging.access.intercept.ChannelSecurityInterceptor.preSend(ChannelSecurityInterceptor.java:72) ~[spring-security-messaging-6.3.4.jar:6.3.4]
        at org.springframework.messaging.support.AbstractMessageChannel$ChannelInterceptorChain.applyPreSend(AbstractMessageChannel.java:181) ~[spring-messaging-6.1.14.jar:6.1.14]
        at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:135) ~[spring-messaging-6.1.14.jar:6.1.14]
        ... 30 common frames omitted

2024-12-02T17:02:24.617+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-3] o.s.messaging.simp.stomp.StompEncoder    : Encoding STOMP ERROR, headers={message=[Failed to send message to ExecutorSubscribableChannel[clientInboundChannel]]}
2024-12-02T17:02:24.618+01:00 TRACE 25252 --- [ecommerce.backend] [nio-8080-exec-3] o.s.w.s.adapter.NativeWebSocketSession   : Sending TextMessage payload=[ERROR 
mess..], byteCount=109, last=true], StandardWebSocketSession[id=59e1bbd2-7469-7e0c-772e-be7aef87d467, uri=ws://localhost:8080/websocket/websocket]
2024-12-02T17:02:24.620+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-3] o.s.w.s.adapter.NativeWebSocketSession   : Closing StandardWebSocketSession[id=59e1bbd2-7469-7e0c-772e-be7aef87d467, uri=ws://localhost:8080/websocket/websocket]
2024-12-02T17:02:24.620+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-3] s.w.s.h.LoggingWebSocketHandlerDecorator : StandardWebSocketSession[id=59e1bbd2-7469-7e0c-772e-be7aef87d467, uri=ws://localhost:8080/websocket/websocket] closed with CloseStatus[code=1002, reason=null]
2024-12-02T17:02:24.620+01:00 DEBUG 25252 --- [ecommerce.backend] [nio-8080-exec-3] o.s.w.s.m.SubProtocolWebSocketHandler    : Clearing session 59e1bbd2-7469-7e0c-772e-be7aef87d467
2024-12-02T17:02:24.621+01:00 DEBUG 25252 --- [ecommerce.backend] [nboundChannel-5] o.s.m.s.b.SimpleBrokerMessageHandler     : Processing DISCONNECT session=59e1bbd2-7469-7e0c-772e-be7aef87d467
2024-12-02T17:02:24.621+01:00  INFO 25252 --- [ecommerce.backend] [nio-8080-exec-3] c.f.e.b.a.s.WebSocketSessionRegistry     : Session removed for user: hello    
2024-12-02T17:02:24.622+01:00 DEBUG 25252 --- [ecommerce.backend] [tboundChannel-2] o.s.w.s.m.SubProtocolWebSocketHandler    : No session for GenericMessage [payload=byte[0], headers={simpMessageType=DISCONNECT_ACK, simpDisconnectMessage=GenericMessage [payload=byte[0], headers={simpMessageType=DISCONNECT, stompCommand=DISCONNECT, simpSessionAttributes={org.springframework.messaging.simp.SimpAttributes.COMPLETED=true}, simpUser=UsernamePasswordAuthenticationToken [Principal=com.foodstore.ecommerce.backend.model.LocalUser@7576a08d, Credentials=[PROTECTED], Authenticated=true, Details=null, Granted Authorities=[ROLE_USER]], simpSessionId=59e1bbd2-7469-7e0c-772e-be7aef87d467}], simpUser=UsernamePasswordAuthenticationToken [Principal=com.foodstore.ecommerce.backend.model.LocalUser@7576a08d, Credentials=[PROTECTED], Authenticated=true, Details=null, Granted Authorities=[ROLE_USER]], simpSessionId=59e1bbd2-7469-7e0c-772e-be7aef87d467}]
2024-12-02T17:03:13.125+01:00  INFO 25252 --- [ecommerce.backend] [MessageBroker-1] o.s.w.s.c.WebSocketMessageBrokerStats    : WebSocketSession[0 current WS(0)-HttpStream(0)-HttpPoll(0), 1 total, 0 closed abnormally (0 connect failure, 0 send limit, 0 transport error)], stompSubProtocol[processed CONNECT(1)-CONNECTED(1)-DISCONNECT(0)], stompBrokerRelay[null], inboundChannel[pool size = 6, active threads = 0, queued tasks = 0, completed tasks = 6], outboundChannel[pool size = 2, active threads = 0, queued tasks = 0, completed tasks = 2], sockJsScheduler[pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]



And here's my code:

package com.foodstore.ecommerce.backend.api.controller;

import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.authority.SimpleGrantedAuthority;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Controller;

import com.foodstore.ecommerce.backend.api.security.ChatMessage;

@Controller
public class WebsocketController {
    private static final Logger logger = LoggerFactory.getLogger(WebsocketController.class);
    private final SimpMessagingTemplate messagingTemplate;

    public WebsocketController(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @MessageMapping("/sendMessage")
    @PreAuthorize("hasRole('ADMIN')")
    public void handleMessage(@Payload ChatMessage message,
            @Header("simpSessionId") String sessionId,
            StompHeaderAccessor headerAccessor) {
        // Get authentication from headerAccessor instead of SecurityContextHolder
        Authentication auth = (Authentication) headerAccessor.getUser();

        if (auth == null) {
            logger.warn("User not authenticated");
            return;
        }

        if (!auth.getAuthorities().stream()
                .anyMatch(a -> a.getAuthority().equals("ROLE_ADMIN"))) {
            logger.warn("Non-admin user attempted to send message");
            return;
        }

        logger.info("Received message - Session: {}, User: {}, Message: {}",
                sessionId,
                auth.getName(),
                message);

        try {
            // Rest of your code remains the same
            if (message.getUserId() == null) {
                logger.error("Invalid message: userId is null");
                return;
            }

            String destination = "/topic/user/" + message.getUserId() + "/address";
            messagingTemplate.convertAndSend(destination, message,
                    createHeaders(headerAccessor.getSessionId()));

            logger.info("Message sent to {}", destination);

        } catch (Exception e) {
            logger.error("Error processing message", e);
        }
    }

    private MessageHeaders createHeaders(String sessionId) {
        Map<String, Object> headers = new HashMap<>();
        headers.put("simpSessionId", sessionId);
        headers.put("content-type", "application/json");
        return new MessageHeaders(headers);
    }

    // Add a test endpoint to verify subscription

    @MessageMapping("/test")
    @SendTo("/topic/test")
    public String handleTest(@Payload(required = false) String message) {
        logger.info("Test endpoint hit: {}", message);
        return "Server received: " + (message != null ? message : "no message");
    }

    @MessageMapping("/admin/message")
    @PreAuthorize("hasRole('ADMIN')") // Add this annotation
    public void handleAdminMessage(@Payload ChatMessage message,
            @Header("simpSessionId") String sessionId,
            StompHeaderAccessor headerAccessor) {
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();

        logger.info("Received admin message - Session: {}, User: {}, Message: {}",
                sessionId,
                auth != null ? auth.getName() : "anonymous",
                message);

        try {
            if (message.getUserId() == null) {
                logger.error("Invalid message: userId is null");
                return;
            }

            // Send to admin-specific topic
            String destination = "/topic/user/" + message.getUserId() + "/admin-updates";
            messagingTemplate.convertAndSend(destination, message,
                    createHeaders(headerAccessor.getSessionId()));

            logger.info("Admin message sent to {}", destination);

        } catch (Exception e) {
            logger.error("Error processing admin message", e);
        }
    }
}



package com.foodstore.ecommerce.backend.api.security;

// DTO Class
public class ChatMessage {
    private Long userId;  // Change to Long instead of String
    private String content;
    
    // Update getters and setters
    public Long getUserId() {
        return userId;
    }

    public void setUserId(Long userId) {
        this.userId = userId;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }
}


package com.foodstore.ecommerce.backend.api.security;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.simp.stomp.*;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.socket.messaging.StompSubProtocolErrorHandler;

import java.nio.charset.StandardCharsets;

public class CustomStompSubProtocolErrorHandler extends StompSubProtocolErrorHandler {

    private static final Logger logger = LoggerFactory.getLogger(CustomStompSubProtocolErrorHandler.class);

    @Override
    public Message<byte[]> handleClientMessageProcessingError(Message<byte[]> clientMessage, Throwable ex) {

        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(clientMessage);
        StompHeaderAccessor errorAccessor = StompHeaderAccessor.create(StompCommand.ERROR);
        errorAccessor.setSessionId(accessor.getSessionId());
        errorAccessor.setLeaveMutable(true);

        // Customize the error message
        String errorText;

        if (ex instanceof org.springframework.security.access.AccessDeniedException) {
            errorText = "Access denied: " + ex.getMessage();
        } else {
            errorText = "An error occurred: " + ex.getMessage();
        }

        errorAccessor.setMessage(errorText);

        logger.error("Handling STOMP error: {}", errorText, ex);

        return MessageBuilder.createMessage(
                errorText.getBytes(StandardCharsets.UTF_8),
                errorAccessor.getMessageHeaders());
    }
}



package com.foodstore.ecommerce.backend.api.security;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.WebSocketHandlerDecorator;

import java.util.Optional;

public class EmaWebSocketHandlerDecorator extends WebSocketHandlerDecorator {

    private static final Logger logger = LoggerFactory.getLogger(EmaWebSocketHandlerDecorator.class);

    public EmaWebSocketHandlerDecorator(WebSocketHandler webSocketHandler) {
        super(webSocketHandler);
    }

    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            WebSocketMessage<?> updatedMessage = updateBodyIfNeeded(textMessage);
            super.handleMessage(session, updatedMessage);
        } else {
            super.handleMessage(session, message);
        }
    }

    private WebSocketMessage<?> updateBodyIfNeeded(TextMessage message) {
        String payload = message.getPayload();

        if (payload.endsWith("\u0000")) {
            return message; // Already formatted correctly
        }

        Optional<StompCommand> stompCommand = getStompCommand(payload);
        if (stompCommand.isPresent() && !stompCommand.get().isBodyAllowed()) {
            if (!payload.endsWith("\n\n")) {
                if (payload.endsWith("\n")) {
                    payload += "\n";
                } else {
                    payload += "\n\n";
                }
            }
        }

        payload += "\u0000"; // Append null character
        return new TextMessage(payload);
    }

    private Optional<StompCommand> getStompCommand(String payload) {
        int firstNewline = payload.indexOf('\n');
        if (firstNewline < 0) {
            return Optional.empty();
        }

        String command = payload.substring(0, firstNewline).trim();
        try {
            return Optional.of(StompCommand.valueOf(command));
        } catch (IllegalArgumentException ex) {
            logger.trace("Error parsing STOMP command from payload: {}", payload, ex);
            return Optional.empty();
        }
    }
}



package com.foodstore.ecommerce.backend.api.security;

// ErrorMessage.java
public class ErrorMessage {
    private String message;
    
    public ErrorMessage(String message) {
        this.message = message;
    }
    
    public String getMessage() {
        return message;
    }
    
    public void setMessage(String message) {
        this.message = message;
    }
}



package com.foodstore.ecommerce.backend.api.security;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.*;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Component
public class ExceptionHandlingChannelInterceptor implements ChannelInterceptor {

    private static final Logger logger = LoggerFactory.getLogger(ExceptionHandlingChannelInterceptor.class);

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        return message;
    }

    @Override
    public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
        if (ex != null) {
            logger.error("Exception in message processing: {}", ex.getMessage(), ex);

            if (ex instanceof AccessDeniedException) {
                StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);

                StompHeaderAccessor errorAccessor = StompHeaderAccessor.create(StompCommand.ERROR);
                errorAccessor.setSessionId(accessor.getSessionId());
                errorAccessor.setLeaveMutable(true);

                String errorText = "Access Denied: " + ex.getMessage();
                errorAccessor.setMessage(errorText);

                // Prevent the connection from closing
                errorAccessor.setHeader("simpCloseConnection", false);

                Message<byte[]> errorMessage = MessageBuilder.createMessage(
                        errorText.getBytes(StandardCharsets.UTF_8),
                        errorAccessor.getMessageHeaders());

                try {
                    channel.send(errorMessage);
                } catch (Exception sendException) {
                    logger.error("Failed to send error message: {}", sendException.getMessage(), sendException);
                }
            }
            // Do not rethrow the exception to prevent connection closure
        }
    }
}




package com.foodstore.ecommerce.backend.api.security;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.GrantedAuthority;
import org.springframework.security.core.authority.SimpleGrantedAuthority;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.web.authentication.WebAuthenticationDetailsSource;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;

import com.auth0.jwt.exceptions.JWTDecodeException;
import com.foodstore.ecommerce.backend.model.LocalUser;
import com.foodstore.ecommerce.backend.model.dao.LocalUserDAO;
import com.foodstore.ecommerce.backend.model.service.JWTService;

import jakarta.servlet.FilterChain;
import jakarta.servlet.ServletException;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;

@Component
public class JWTRequestFilter extends OncePerRequestFilter implements ChannelInterceptor {

    private static final Logger logger = LoggerFactory.getLogger(JWTRequestFilter.class);
    private JWTService jwtService;
    private LocalUserDAO localUserDAO;

    public JWTRequestFilter(JWTService jwtService, LocalUserDAO localUserDAO) {
        this.jwtService = jwtService;
        this.localUserDAO = localUserDAO;
    }

    @Override
    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
            throws ServletException, IOException {
        String tokenHeader = request.getHeader("Authorization");
        processToken(tokenHeader);
        filterChain.doFilter(request, response);
    }

    private void processToken(String tokenHeader) {
        if (tokenHeader != null && tokenHeader.startsWith("Bearer ")) {
            String token = tokenHeader.substring(7);
            try {
                String username = jwtService.getUsername(token);
                Optional<LocalUser> opUser = localUserDAO.findByUsernameIgnoreCase(username);
                if (opUser.isPresent()) {
                    LocalUser user = opUser.get();
                    if (user.isEmailVerified()) {
                        List<GrantedAuthority> authorities = List.of(
                                new SimpleGrantedAuthority("ROLE_" + user.getRole().name()));
                        UsernamePasswordAuthenticationToken authToken = new UsernamePasswordAuthenticationToken(
                                user, null, authorities);
                        SecurityContextHolder.getContext().setAuthentication(authToken);
                    }
                }
            } catch (Exception e) {
                logger.error("Error processing JWT token", e);
            }
        }
    }

    UsernamePasswordAuthenticationToken checkToken(String token) {
        if (token != null && token.startsWith("Bearer ")) {
            token = token.substring(7);
            try {
                String username = jwtService.getUsername(token);
                Optional<LocalUser> opUser = localUserDAO.findByUsernameIgnoreCase(username);
                if (opUser.isPresent()) {
                    LocalUser user = opUser.get();
                    if (user.isEmailVerified()) {
                        return new UsernamePasswordAuthenticationToken(
                                user, null, List.of(new SimpleGrantedAuthority("ROLE_" + user.getRole().name())));
                    }
                }
            } catch (JWTDecodeException ex) {
                System.err.println("Invalid token: " + token);
                ex.printStackTrace();
            }
        }
        return null;
    }

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);

        if (accessor != null &&
                (StompCommand.CONNECT.equals(accessor.getCommand()) ||
                        StompCommand.SEND.equals(accessor.getCommand()))) {

            List<String> authorization = accessor.getNativeHeader("Authorization");
            if (authorization != null && !authorization.isEmpty()) {
                String tokenHeader = authorization.get(0);
                Authentication auth = checkToken(tokenHeader);
                if (auth != null) {
                    accessor.setUser(auth);
                    // Set user attributes in session attributes for future reference
                    accessor.setSessionAttributes(Map.of("user", auth.getPrincipal()));
                }
            }
        }

        return message;
    }
}




package com.foodstore.ecommerce.backend.api.security;

import java.util.regex.Pattern;

import org.springframework.stereotype.Component;

// MessageValidator.java
@Component
public class MessageValidator {
    private static final int MAX_CONTENT_LENGTH = 1000;
    private static final Pattern CONTENT_PATTERN = Pattern.compile("^[\\w\\s.,!?-]+$");

    public void validate(ChatMessage message) throws ValidationException {
        if (message == null) {
            throw new ValidationException("Message cannot be null");
        }
        if (message.getUserId() == null || message.getUserId() <= 0) {
            throw new ValidationException("Invalid userId");
        }
        if (message.getContent() == null) {
            throw new ValidationException("Message content cannot be null");
        }
        String content = message.getContent().trim();
        if (content.isEmpty()) {
            throw new ValidationException("Message content cannot be empty");
        }
        if (content.length() > MAX_CONTENT_LENGTH) {
            throw new ValidationException("Message content exceeds maximum length of " + MAX_CONTENT_LENGTH);
        }
        if (!CONTENT_PATTERN.matcher(content).matches()) {
            throw new ValidationException("Message content contains invalid characters");
        }
    }
}



package com.foodstore.ecommerce.backend.api.security;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

// RateLimiter.java
@Component
public class RateLimiter {
    private final Map<String, TokenBucket> buckets = new ConcurrentHashMap<>();
    private static final int NORMAL_USER_CAPACITY = 10;
    private static final int ADMIN_CAPACITY = 50;
    private static final int REFILL_TOKENS = 1;
    private static final int REFILL_TIME_MS = 1000;

    @Scheduled(fixedRate = 1000)
    public void refillBuckets() {
        buckets.values().forEach(bucket -> bucket.refill(REFILL_TOKENS));
    }

    public boolean tryAcquire(String sessionId, boolean isAdmin) {
        TokenBucket bucket = buckets.computeIfAbsent(sessionId,
                k -> new TokenBucket(isAdmin ? ADMIN_CAPACITY : NORMAL_USER_CAPACITY));
        return bucket.tryConsume();
    }

    private static class TokenBucket {
        private int tokens;
        private final int capacity;

        public TokenBucket(int capacity) {
            this.capacity = capacity;
            this.tokens = capacity;
        }

        public synchronized boolean tryConsume() {
            if (tokens > 0) {
                tokens--;
                return true;
            }
            return false;
        }

        public synchronized void refill(int count) {
            tokens = Math.min(capacity, tokens + count);
        }
    }
}





package com.foodstore.ecommerce.backend.api.security;

// ValidationException.java
public class ValidationException extends RuntimeException {
    public ValidationException(String message) {
        super(message);
    }
}



package com.foodstore.ecommerce.backend.api.security;

import java.util.Arrays;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.security.config.annotation.web.builders.HttpSecurity;
import org.springframework.security.web.SecurityFilterChain;
import org.springframework.security.web.access.intercept.AuthorizationFilter;
import org.springframework.stereotype.Controller;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.cors.CorsConfigurationSource;
import org.springframework.web.cors.UrlBasedCorsConfigurationSource;

@Configuration
public class WebSecurityConfig {
    private JWTRequestFilter jwtRequestFilter;

    public WebSecurityConfig(JWTRequestFilter jwtRequestFilter) {
        this.jwtRequestFilter = jwtRequestFilter;
    }

    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
        http
                .csrf().disable()
                .cors() // Enable CORS (remove the .disable())
                .and()
                .headers().cacheControl().disable()
                .and()
                .addFilterBefore(jwtRequestFilter, AuthorizationFilter.class)
                .authorizeHttpRequests()
                .requestMatchers("/", "/product", "/auth/register", "/auth/login", "/auth/verify", "/error",
                        "/auth/forgot",
                        "/auth/reset", "/websocket", "/websocket/**",
                        "/websocket",
                        "/websocket/info",
                        "/websocket/websocket/**", "/images/**", "/placeholder-image.jpg")
                .permitAll().requestMatchers("/product/admin/**").hasRole("ADMIN")
                .anyRequest().authenticated();
        return http.build();
    }

    @Bean
    public CorsConfigurationSource corsConfigurationSource() {
        CorsConfiguration configuration = new CorsConfiguration();
        configuration.setAllowedOrigins(Arrays.asList("http://localhost:3000"));
        configuration.setAllowedMethods(Arrays.asList("GET", "POST", "PUT", "DELETE", "OPTIONS"));
        configuration.setAllowedHeaders(Arrays.asList(
                "Authorization",
                "Content-Type",
                "Accept",
                "x-requested-with",
                "x-auth-token"));
        configuration.setAllowCredentials(true);
        configuration.setMaxAge(3600L);

        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
        source.registerCorsConfiguration("/**", configuration);
        return source;
    }

    @Controller
    public class WebSocketController {
        @MessageMapping("/profile")
        @SendTo("/topic/profile")
        public String handleProfile(String message) {
            System.out.println("Message received at /profile: " + message);
            return "Response from /profile: " + message;
        }
    }

}




package com.foodstore.ecommerce.backend.api.security;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Component;

// WebSocketChannelInterceptor.java (update your existing one)
@Component
public class WebSocketChannelInterceptor implements ChannelInterceptor {

    private final RateLimiter rateLimiter;
    private final MessageValidator messageValidator;
    private static final Logger logger = LoggerFactory.getLogger(WebSocketChannelInterceptor.class);

    public WebSocketChannelInterceptor(RateLimiter rateLimiter, MessageValidator messageValidator) {
        this.rateLimiter = rateLimiter;
        this.messageValidator = messageValidator;
    }

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);

        if (accessor != null) {
            try {
                if (accessor.getCommand() == null) {
                    return message;
                }

                switch (accessor.getCommand()) {
                    case CONNECT:
                        logger.debug("Processing CONNECT command");
                        return handleConnect(message, accessor);
                    case SUBSCRIBE:
                        logger.debug("Processing SUBSCRIBE command");
                        return handleSubscribe(message, accessor);
                    case SEND:
                        logger.debug("Processing SEND command");
                        return handleMessageSend(message, accessor);
                    default:
                        return message;
                }
            } catch (Exception e) {
                logger.error("Exception in WebSocketChannelInterceptor: {}", e.getMessage(), e);
                return createErrorMessage(accessor, "Internal server error: " + e.getMessage());
            }
        }
        return message;
    }

    private Message<?> handleConnect(Message<?> message, StompHeaderAccessor accessor) {
        Authentication auth = (Authentication) accessor.getUser();
        if (auth == null) {
            logger.warn("Connection attempt without authentication");
            return createErrorMessage(accessor, "Connection attempt without authentication");
        }
        logger.info("User {} connected", auth.getName());
        return message;
    }

    private Message<?> handleSubscribe(Message<?> message, StompHeaderAccessor accessor) {
        Authentication auth = (Authentication) accessor.getUser();
        String destination = accessor.getDestination();

        if (auth == null) {
            logger.warn("Subscription attempt without authentication");
            return createErrorMessage(accessor, "Subscription attempt without authentication");
        }

        if (destination != null) {
            logger.info("User {} subscribing to {}", auth.getName(), destination);
            // You can add rate limiting or validation logic here if needed
        }

        return message;
    }

    private Message<?> handleMessageSend(Message<?> message, StompHeaderAccessor accessor) {
        Authentication auth = (Authentication) accessor.getUser();
        if (auth == null) {
            logger.warn("Message send attempt without authentication");
            return createErrorMessage(accessor, "Message send attempt without authentication");
        }

        // Validate message if it's a ChatMessage
        if (message.getPayload() instanceof ChatMessage) {
            try {
                messageValidator.validate((ChatMessage) message.getPayload());
            } catch (ValidationException e) {
                logger.warn("Message validation failed for user {}: {}", auth.getName(), e.getMessage());
                return createErrorMessage(accessor, "Invalid message content: " + e.getMessage());
            }
        }

        return message;
    }

    private Message<?> createErrorMessage(StompHeaderAccessor accessor, String errorText) {
        StompHeaderAccessor errorAccessor = StompHeaderAccessor.create(StompCommand.ERROR);
        errorAccessor.setMessage(errorText);
        errorAccessor.setSessionId(accessor.getSessionId());
        errorAccessor.setUser(accessor.getUser());

        return MessageBuilder.createMessage(new byte[0], errorAccessor.getMessageHeaders());
    }

    @Override
    public void afterSendCompletion(
            Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
        if (ex != null) {
            logger.error("Exception in WebSocketChannelInterceptor afterSendCompletion: {}", ex.getMessage(), ex);
            // Handle the exception as needed
        }
    }
}



package com.foodstore.ecommerce.backend.api.security;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.security.authorization.AuthorizationEventPublisher;
import org.springframework.security.authorization.AuthorizationManager;
import org.springframework.security.authorization.SpringAuthorizationEventPublisher;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.authority.SimpleGrantedAuthority;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.messaging.access.intercept.AuthorizationChannelInterceptor;
import org.springframework.security.messaging.access.intercept.MessageMatcherDelegatingAuthorizationManager;
import org.springframework.stereotype.Component;
import org.springframework.util.AntPathMatcher;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;
import org.springframework.web.socket.handler.WebSocketHandlerDecorator;
import org.springframework.web.socket.messaging.StompSubProtocolErrorHandler;

import com.foodstore.ecommerce.backend.model.LocalUser;
import com.foodstore.ecommerce.backend.model.service.UserService;

import jakarta.validation.ValidationException;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;

import com.foodstore.ecommerce.backend.api.security.ErrorMessage;
import com.foodstore.ecommerce.backend.api.security.WebSocketSessionRegistry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.socket.config.annotation.*;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.config.annotation.*;
import org.springframework.web.socket.handler.WebSocketHandlerDecorator;

@Configuration
@EnableWebSocket
@EnableWebSocketMessageBroker
public class WebsocketConfiguration implements WebSocketMessageBrokerConfigurer {

    private static final Logger logger = LoggerFactory.getLogger(WebsocketConfiguration.class);
    private final ApplicationContext context;
    private final JWTRequestFilter jwtRequestFilter;
    private final UserService userService;
    private final SimpMessagingTemplate messagingTemplate;
    private final RateLimiter rateLimiter;
    private final MessageValidator messageValidator;
    private final WebSocketSessionRegistry sessionRegistry;
    private static final AntPathMatcher MATCHER = new AntPathMatcher();
    private final WebSocketSecurityInterceptor securityInterceptor;
    private final WebSocketChannelInterceptor channelInterceptor;
    private ExceptionHandlingChannelInterceptor exceptionHandlingChannelInterceptor;

    public WebsocketConfiguration(
            ApplicationContext context,
            JWTRequestFilter jwtRequestFilter,
            UserService userService,
            @Lazy SimpMessagingTemplate messagingTemplate,
            RateLimiter rateLimiter,
            @Lazy MessageValidator messageValidator,
            @Lazy WebSocketSessionRegistry sessionRegistry,
            WebSocketSecurityInterceptor securityInterceptor,
            WebSocketChannelInterceptor channelInterceptor,
            ExceptionHandlingChannelInterceptor exceptionHandlingChannelInterceptor) {
        this.context = context;
        this.jwtRequestFilter = jwtRequestFilter;
        this.userService = userService;
        this.messagingTemplate = messagingTemplate;
        this.rateLimiter = rateLimiter;
        this.messageValidator = messageValidator;
        this.sessionRegistry = sessionRegistry;
        this.securityInterceptor = securityInterceptor;
        this.channelInterceptor = channelInterceptor;
        this.exceptionHandlingChannelInterceptor = exceptionHandlingChannelInterceptor;
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/websocket")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.setApplicationDestinationPrefixes("/app");
        registry.enableSimpleBroker("/topic", "/queue")
                .setHeartbeatValue(new long[] { 25000, 25000 })
                .setTaskScheduler(heartBeatScheduler());
    }

    private void validateSubscription(StompHeaderAccessor accessor) {
        String destination = accessor.getDestination();
        if (destination == null) {
            throw new ValidationException("Subscription destination cannot be null");
        }

        if (!destination.startsWith("/topic/user/")) {
            throw new ValidationException("Invalid subscription destination");
        }

        Authentication auth = (Authentication) accessor.getUser();
        if (auth == null) {
            throw new ValidationException("User not authenticated");
        }

        try {
            String[] parts = destination.split("/");
            Long userId = Long.parseLong(parts[3]);
            LocalUser user = (LocalUser) auth.getPrincipal();
            if (!user.getId().equals(userId)
                    && !auth.getAuthorities().contains(new SimpleGrantedAuthority("ROLE_ADMIN"))) {
                throw new ValidationException("Unauthorized subscription");
            }
        } catch (Exception e) {
            throw new ValidationException("Invalid subscription format");
        }
    }

    // Add reconnection strategy
    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
        registration
                .setMessageSizeLimit(128 * 1024)
                .setSendTimeLimit(20 * 1000)
                .setSendBufferSizeLimit(512 * 1024)
                .addDecoratorFactory(handler -> new WebSocketHandlerDecorator(handler) {
                    @Override
                    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
                        super.afterConnectionEstablished(session);
                        sessionRegistry.registerSession(session);
                    }

                    @Override
                    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
                        super.afterConnectionClosed(session, status);
                        sessionRegistry.removeSession(session.getId());
                    }
                })
                .addDecoratorFactory(webSocketHandler -> new EmaWebSocketHandlerDecorator(webSocketHandler));
                 // Register custom error handler
    }

    @Bean
    public StompSubProtocolErrorHandler customStompSubProtocolErrorHandler() {
        return new CustomStompSubProtocolErrorHandler();
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        ThreadPoolTaskExecutor clientInboundExecutor = new ThreadPoolTaskExecutor();
        clientInboundExecutor.setCorePoolSize(10);
        clientInboundExecutor.setMaxPoolSize(100);
        clientInboundExecutor.setQueueCapacity(500);
        clientInboundExecutor.setKeepAliveSeconds(60);
        clientInboundExecutor.setThreadNamePrefix("clientInboundChannel-");
        clientInboundExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        clientInboundExecutor.initialize();

        registration.taskExecutor(clientInboundExecutor);

        // Add interceptors in the correct order
        registration.interceptors(
                exceptionHandlingChannelInterceptor, // Must be first
                securityInterceptor,
                channelInterceptor
        );
    }

    private Authentication checkAuthToken(String bearerToken) {
        String token = bearerToken.startsWith("Bearer ") ? bearerToken.substring(7) : bearerToken;
        try {
            return jwtRequestFilter.checkToken(token);
        } catch (Exception e) {
            logger.error("Error validating WebSocket token", e);
            return null;
        }
    }

    private class RejectClientMessagesOnChannelsChannelInterceptor implements ChannelInterceptor {
        private final String[] protectedPaths = {
                "/topic/user/*/address"
        };

        @Override
        public Message<?> preSend(Message<?> message, MessageChannel channel) {
            StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
            if (accessor != null && accessor.getMessageType() == SimpMessageType.MESSAGE) {
                String destination = accessor.getDestination();
                if (destination != null) {
                    for (String path : protectedPaths) {
                        if (MATCHER.match(path, destination)) {
                            logger.warn("Rejected client message to protected destination: {}", destination);
                            return null;
                        }
                    }
                }
            }
            return message;
        }
    }

    private class DestinationLevelAuthorizationChannelInterceptor implements ChannelInterceptor {
        @Override
        public Message<?> preSend(Message<?> message, MessageChannel channel) {
            StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
            if (accessor != null && accessor.getMessageType() == SimpMessageType.SUBSCRIBE) {
                String destination = accessor.getDestination();
                if (destination != null) {
                    String userTopicMatcher = "/topic/user/{userId}/**";
                    if (MATCHER.match(userTopicMatcher, destination)) {
                        logger.info("Processing subscription request to: {}", destination);
                        // Add any additional subscription authorization logic here if needed
                    }
                }
            }
            return message;
        }
    }

    private AuthorizationManager<Message<?>> makeMessageAuthorizationManager() {
        MessageMatcherDelegatingAuthorizationManager.Builder messages = new MessageMatcherDelegatingAuthorizationManager.Builder();
        messages
                .simpDestMatchers("/topic/user/**").authenticated()
                .simpDestMatchers("/app/**").authenticated()
                .anyMessage().permitAll();
        return messages.build();
    }

    @Bean
    public TaskScheduler websocketHeartbeatScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(1);
        scheduler.setThreadNamePrefix("websocket-heartbeat-thread-");
        scheduler.setDaemon(true);
        scheduler.initialize();
        return scheduler;
    }

    // Register the custom error handler
    // Register the custom error handler

    @Bean
    public TaskScheduler heartBeatScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(1);
        scheduler.setThreadNamePrefix("ws-heartbeat-");
        scheduler.setDaemon(true);
        return scheduler;
    }
}



package com.foodstore.ecommerce.backend.api.security;

import java.security.Principal;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.ControllerAdvice;

@ControllerAdvice
public class WebSocketErrorHandler {

    private final SimpMessagingTemplate messagingTemplate;
    private static final Logger logger = LoggerFactory.getLogger(WebSocketErrorHandler.class);

    @Autowired
    public WebSocketErrorHandler(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @MessageExceptionHandler(AccessDeniedException.class)
    public void handleAccessDeniedException(AccessDeniedException exception, Principal principal) {
        String errorMessage = "Access denied: " + exception.getMessage();
        logger.error(errorMessage, exception);

        if (principal != null) {
            messagingTemplate.convertAndSendToUser(
                    principal.getName(),
                    "/queue/errors",
                    new ErrorMessage(errorMessage));
        }
    }

    @MessageExceptionHandler(Exception.class)
    public void handleException(Exception exception, Principal principal) {
        String errorMessage = "Error processing message: " + exception.getMessage();
        logger.error(errorMessage, exception);

        if (principal != null) {
            messagingTemplate.convertAndSendToUser(
                    principal.getName(),
                    "/queue/errors",
                    new ErrorMessage(errorMessage));
        }
    }
}




package com.foodstore.ecommerce.backend.api.security;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.web.messaging.MessageSecurityMetadataSourceRegistry;
import org.springframework.security.config.annotation.web.socket.AbstractSecurityWebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketSecurityConfig extends AbstractSecurityWebSocketMessageBrokerConfigurer {

    @Autowired
    private WebSocketSecurityService webSocketSecurityService;

    @Override
    protected void configureInbound(MessageSecurityMetadataSourceRegistry messages) {
        messages
                .simpDestMatchers("/topic/admin/**").hasRole("ADMIN")
                .simpDestMatchers("/topic/user/{userId}/admin-updates/**").hasRole("ADMIN")
                .simpDestMatchers("/topic/user/{userId}/**")
                .access("@webSocketSecurityService.checkUserId(authentication, #userId)")
                .simpDestMatchers("/topic/public/**").authenticated()
                .simpDestMatchers("/app/**").authenticated()
                .anyMessage().authenticated();
    }

    @Override
    protected boolean sameOriginDisabled() {
        return true; // Disable CSRF for WebSocket endpoints
    }
}




package com.foodstore.ecommerce.backend.api.security;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.security.core.Authentication;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.AntPathMatcher;

@Component
public class WebSocketSecurityInterceptor implements ChannelInterceptor {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketSecurityInterceptor.class);

    private static final AntPathMatcher matcher = new AntPathMatcher();

    @Autowired
    private WebSocketSecurityService webSocketSecurityService;

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {

        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);

        if (accessor != null) {
            if (accessor.getCommand() == null) {
                return message;
            }

            switch (accessor.getCommand()) {
                case SUBSCRIBE:
                    return handleSubscribe(message, accessor);
                case SEND:
                    return handleSend(message, accessor);
                default:
                    return message;
            }
        }

        return message;
    }

    private Message<?> handleSubscribe(Message<?> message, StompHeaderAccessor accessor) {
        Authentication authentication = (Authentication) accessor.getUser();
        String destination = accessor.getDestination();

        if (authentication != null && destination != null) {
            String userId = extractUserIdFromDestination(destination);
            if (userId == null) {
                logger.warn("Invalid destination format: {}", destination);
                return createErrorMessage(accessor, "Invalid destination format.");
            }
            if (!webSocketSecurityService.checkUserId(authentication, userId)) {
                logger.warn("Access denied for user: {}", authentication.getName());
                return createErrorMessage(accessor, "Access denied: insufficient permissions.");
            }
        } else {
            logger.warn("Unauthenticated subscription attempt");
            return createErrorMessage(accessor, "Unauthenticated subscription attempt.");
        }

        return message; // Continue processing
    }

    private Message<?> handleSend(Message<?> message, StompHeaderAccessor accessor) {
        Authentication authentication = (Authentication) accessor.getUser();

        if (authentication != null) {
            boolean isAdmin = authentication.getAuthorities().stream()
                    .anyMatch(auth -> auth.getAuthority().equals("ROLE_ADMIN"));

            if (!isAdmin) {
                logger.warn("Access denied for user: {}", authentication.getName());
                return createErrorMessage(accessor, "Access denied: insufficient permissions.");
            }
        } else {
            logger.warn("Unauthenticated message send attempt");
            return createErrorMessage(accessor, "Unauthenticated message send attempt.");
        }

        return message;
    }

    private String extractUserIdFromDestination(String destination) {
        String pattern = "/topic/user/{userId}/**";

        if (matcher.match(pattern, destination)) {
            Map<String, String> variables = matcher.extractUriTemplateVariables(pattern, destination);
            return variables.get("userId");
        } else {
            logger.warn("Destination does not match pattern: {}", destination);
            return null; // Or return an error message
        }
    }

    private Message<?> createErrorMessage(StompHeaderAccessor accessor, String errorText) {
        StompHeaderAccessor errorAccessor = StompHeaderAccessor.create(StompCommand.ERROR);
        errorAccessor.setSessionId(accessor.getSessionId());
        errorAccessor.setUser(accessor.getUser());
        errorAccessor.setMessage(errorText);

        return MessageBuilder.createMessage(new byte[0], errorAccessor.getMessageHeaders());
    }
}




package com.foodstore.ecommerce.backend.api.security;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Service;

import com.foodstore.ecommerce.backend.model.LocalUser;

// WebSocketSecurityService.java
@Service
public class WebSocketSecurityService {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketSecurityService.class);

    public boolean checkUserId(Authentication authentication, String userId) {
        if (authentication == null || !(authentication.getPrincipal() instanceof LocalUser)) {
            logger.warn("Invalid authentication or principal");
            return false;
        }

        LocalUser user = (LocalUser) authentication.getPrincipal();

        // Allow admins to access all channels
        if (user.getRole() == LocalUser.UserRole.ADMIN) {
            logger.debug("Admin access granted for user {}", user.getUsername());
            return true;
        }

        // Allow users to access only their own channels
        boolean hasAccess = user.getId().toString().equals(userId);
        logger.debug("User {} access to channel {}: {}", user.getUsername(), userId, hasAccess);
        return hasAccess;
    }
}




package com.foodstore.ecommerce.backend.api.security;

import java.security.Principal;
import java.util.Collections;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

// WebSocketSessionRegistry.java
@Component
public class WebSocketSessionRegistry {
    private final ConcurrentHashMap<String, WebSocketSession> sessions = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Set<String>> userSessions = new ConcurrentHashMap<>();
    private static final Logger logger = LoggerFactory.getLogger(WebSocketSessionRegistry.class);

    public void registerSession(WebSocketSession session) {
        sessions.put(session.getId(), session);
        Principal principal = session.getPrincipal();
        if (principal != null) {
            userSessions.computeIfAbsent(principal.getName(), k -> ConcurrentHashMap.newKeySet())
                       .add(session.getId());
            logger.info("Session registered for user: {}", principal.getName());
        }
    }

    public void removeSession(String sessionId) {
        WebSocketSession session = sessions.remove(sessionId);
        if (session != null && session.getPrincipal() != null) {
            String username = session.getPrincipal().getName();
            Set<String> userSessionSet = userSessions.get(username);
            if (userSessionSet != null) {
                userSessionSet.remove(sessionId);
                if (userSessionSet.isEmpty()) {
                    userSessions.remove(username);
                }
            }
            logger.info("Session removed for user: {}", username);
        }
    }

    public Set<WebSocketSession> getUserSessions(String username) {
        Set<String> sessionIds = userSessions.getOrDefault(username, Collections.emptySet());
        return sessionIds.stream()
                        .map(sessions::get)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toSet());
    }
}




Please don't change anything else, just fix the issue so that it doesn't get disconnected and instead shows an error message when a regular user tries to subscribe to websockets topic with a different userid