RED-9352: Changed to complete communication via websocket queue

build.gradle.kts

@@ -2,8 +2,8 @@ import org.springframework.boot.gradle.tasks.bundling.BootBuildImage
 
 plugins {
     java
-    id("org.springframework.boot") version "3.3.1"
+    id("org.springframework.boot") version "3.1.5"
-    id("io.spring.dependency-management") version "1.1.5"
+    id("io.spring.dependency-management") version "1.1.4"
     id("org.sonarqube") version "4.4.1.3373"
     id("io.freefair.lombok") version "8.6"
     pmd
@@ -92,7 +92,7 @@ configurations {
 }
 
 extra["springCloudVersion"] = "2022.0.5"
-extra["testcontainersVersion"] = "1.19.8"
+extra["testcontainersVersion"] = "1.19.7"
 
 dependencies {
     implementation("org.springframework.boot:spring-boot-starter-actuator")
@@ -100,7 +100,7 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-web")
     implementation("org.springframework.cloud:spring-cloud-starter-openfeign")
     implementation("org.springframework.boot:spring-boot-starter-websocket")
-    implementation("org.springframework.security:spring-security-messaging:6.3.1")
+    implementation("org.springframework.security:spring-security-messaging:6.1.3")
     implementation("com.iqser.red.commons:storage-commons:2.49.0")
     implementation("com.knecon.fforesight:keycloak-commons:0.29.0")
     implementation("com.knecon.fforesight:swagger-commons:0.7.0")

publish-custom-image.sh

@@ -4,5 +4,5 @@ gradle assemble
 
 buildNumber=${1:-1}
 
-gradle bootBuildImage --cleanCache --publishImage -PbuildbootDockerHostNetwork=true -Pversion=$USER-$buildNumber
+gradle bootBuildImage --cleanCache --publishImage -Pversion=$USER-$buildNumber
 echo "nexus.knecon.com:5001/red/${dir}-server-v1:$USER-$buildNumber"

src/main/java/com/knecon/fforesight/llm/service/controller/external/LlmContoller.java

@@ -1,43 +0,0 @@
-package com.knecon.fforesight.llm.service.controller.external;
-
-import org.springframework.http.MediaType;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.ResponseBody;
-import org.springframework.web.bind.annotation.RestController;
-
-import com.knecon.fforesight.keycloakcommons.security.KeycloakSecurity;
-import com.knecon.fforesight.llm.service.api.model.PromptList;
-import com.knecon.fforesight.llm.service.services.LlmService;
-
-import io.swagger.v3.oas.annotations.Operation;
-import io.swagger.v3.oas.annotations.responses.ApiResponse;
-import io.swagger.v3.oas.annotations.responses.ApiResponses;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import io.swagger.v3.oas.annotations.tags.Tags;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@RestController
-@RequiredArgsConstructor
-@RequestMapping("${fforesight.llm-service.base-path:/api/llm}")
-@Tags(value = {@Tag(name = "LLM", description = "Provides endpoint to call llm")})
-public class LlmContoller {
-
-    private final LlmService llmService;
-
-
-    @ResponseBody
-//    @PreAuthorize("hasAuthority('" + LLM + "')")
-    @Operation(summary = "Make a request to the llm, response will be streamed to websocket.")
-    @ApiResponses(value = {@ApiResponse(responseCode = "200"), @ApiResponse(responseCode = "400", description = "Bad Request. Something is not right in the request object. See response for details."), @ApiResponse(responseCode = "401", description = "Unauthorized."), @ApiResponse(responseCode = "403", description = "Forbidden.")})
-    @PostMapping(value = "/chat-async", consumes = MediaType.APPLICATION_JSON_VALUE)
-    public void chat(@RequestBody PromptList promptList) {
-
-        log.info("UserId is: {}", KeycloakSecurity.getUserId());
-        llmService.execute(promptList.getPrompts());
-    }
-
-}

src/main/java/com/knecon/fforesight/llm/service/controller/external/LlmController.java

@@ -0,0 +1,18 @@
+package com.knecon.fforesight.llm.service.controller.external;
+
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import io.swagger.v3.oas.annotations.tags.Tag;
+import io.swagger.v3.oas.annotations.tags.Tags;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@RestController
+@RequiredArgsConstructor
+@RequestMapping("${fforesight.llm-service.base-path:/api/llm}")
+@Tags(value = {@Tag(name = "LLM", description = "Provides endpoint to call llm")})
+public class LlmController {
+
+}

src/main/java/com/knecon/fforesight/llm/service/controller/external/WebsocketController.java

@@ -0,0 +1,27 @@
+package com.knecon.fforesight.llm.service.controller.external;
+
+import java.security.Principal;
+
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.stereotype.Controller;
+
+import com.knecon.fforesight.llm.service.api.model.PromptList;
+import com.knecon.fforesight.llm.service.services.LlmService;
+
+import lombok.RequiredArgsConstructor;
+
+@Controller
+@RequiredArgsConstructor
+public class WebsocketController {
+
+    private final LlmService llmService;
+
+
+    @MessageMapping("/rules-copilot")
+    public void rulesCopilot(@Payload PromptList promptList, Principal user) {
+
+        llmService.rulesCopilot(promptList.getPrompts(), user.getName());
+    }
+
+}

src/main/java/com/knecon/fforesight/llm/service/services/LlmService.java

@@ -13,7 +13,6 @@ import com.azure.ai.openai.models.ChatCompletionsOptions;
 import com.azure.ai.openai.models.ChatMessage;
 import com.azure.ai.openai.models.ChatRole;
 import com.azure.core.credential.AzureKeyCredential;
-import com.knecon.fforesight.keycloakcommons.security.KeycloakSecurity;
 import com.knecon.fforesight.llm.service.api.model.ChatEvent;
 import com.knecon.fforesight.llm.service.model.SystemMessages;
 import com.knecon.fforesight.llm.service.settings.LlmServiceSettings;
@@ -43,7 +42,7 @@ public class LlmService {
 
 
     @SneakyThrows
-    public void execute(List<String> prompt) {
+    public void rulesCopilot(List<String> prompt, String userId) {
 
         List<ChatMessage> chatMessages = new ArrayList<>();
         chatMessages.add(new ChatMessage(ChatRole.SYSTEM, SystemMessages.RULES_CO_PILOT));
@@ -53,9 +52,9 @@ public class LlmService {
         ChatCompletionsOptions options = new ChatCompletionsOptions(chatMessages);
         options.setStream(true);
         Flux<ChatCompletions> chatCompletions = client.getChatCompletionsStream(settings.getModel(), options);
-        String userId = KeycloakSecurity.getUserId();
         chatCompletions.subscribe(chatCompletion -> {
-            sendWebsocketEvent(userId, chatCompletion.getChoices()
+            sendWebsocketEvent(userId,
+                               chatCompletion.getChoices()
                                        .get(0).getDelta().getContent());
 
         });
@@ -64,7 +63,7 @@ public class LlmService {
 
     private void sendWebsocketEvent(String userId, String token) {
 
-        websocketTemplate.convertAndSend("/topic/" + TenantContext.getTenantId() + "/chat-events/" + userId, new ChatEvent(token));
+        websocketTemplate.convertAndSendToUser(userId, "/queue/" + TenantContext.getTenantId() + "/rules-copilot", new ChatEvent(token));
     }
 
 }

src/main/java/com/knecon/fforesight/llm/service/websocket/WebSocketConfiguration.java

@@ -41,8 +41,9 @@ public class WebSocketConfiguration implements WebSocketMessageBrokerConfigurer
     public void configureMessageBroker(MessageBrokerRegistry config) {
 
         config.setPreservePublishOrder(true);
-        config.enableSimpleBroker("/topic");
+        config.enableSimpleBroker("/topic", "/queue", "/user");
         config.setApplicationDestinationPrefixes("/app");
+        config.setUserDestinationPrefix("/user");
     }
 
 

src/main/java/com/knecon/fforesight/llm/service/websocket/WebSocketSecurityConfig.java

@@ -13,6 +13,7 @@ import org.springframework.security.config.annotation.web.socket.AbstractSecurit
 import org.springframework.security.oauth2.server.resource.authentication.JwtAuthenticationToken;
 
 import com.knecon.fforesight.keycloakcommons.security.TokenUtils;
+import com.knecon.fforesight.tenantcommons.TenantContext;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -34,7 +35,7 @@ public class WebSocketSecurityConfig extends AbstractSecurityWebSocketMessageBro
                 .simpTypeMatchers(SimpMessageType.SUBSCRIBE)
                 .access("@tenantWebSocketSecurityMatcher.checkCanSubscribeTo(authentication,message)")
                 .anyMessage()
-                .denyAll();
+                .access("@tenantWebSocketSecurityMatcher.checkAuthAndSetTenant(authentication,message)");
     }
 
 
@@ -62,6 +63,16 @@ public class WebSocketSecurityConfig extends AbstractSecurityWebSocketMessageBro
         }
 
 
+        public boolean checkAuthAndSetTenant(JwtAuthenticationToken authentication, Message<?> message){
+            Optional<String> tenantId = getCurrentTenant(authentication);
+            if (tenantId.isPresent()){
+                TenantContext.setTenantId(tenantId.get());
+                return true;
+            }
+            return false;
+        }
+
+
         private Optional<String> getCurrentTenant(JwtAuthenticationToken authentication) {
 
             if (authentication != null && authentication.getToken() != null && authentication.getToken().getTokenValue() != null) {
@@ -76,13 +87,18 @@ public class WebSocketSecurityConfig extends AbstractSecurityWebSocketMessageBro
     private Optional<String> extractTenantId(Message<?> message) {
 
         StompHeaderAccessor sha = StompHeaderAccessor.wrap(message);
-        String topic = sha.getDestination();
+        String destination = sha.getDestination();
-        if (topic == null) {
+        if (destination == null) {
             return Optional.empty();
         }
 
-        String tenant = topic.split("/")[2];
+        if (destination.contains("topic")) {
-        return Optional.of(tenant);
+            return Optional.of(destination.split("topic/")[1].split("/")[0]);
+        } else if (destination.contains("queue")) {
+            return Optional.of(destination.split("queue/")[1].split("/")[0]);
+        }
+
+        return Optional.empty();
     }
 
 }

src/test/java/com/knecon/fforesight/llm/service/LlmServiceIntegrationTest.java

@@ -1,6 +1,6 @@
 package com.knecon.fforesight.llm.service;
 
-import java.util.List;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
@@ -9,14 +9,16 @@ import org.springframework.beans.factory.annotation.Autowired;
 import com.knecon.fforesight.llm.service.services.LlmService;
 
 @Disabled
-public class LlmServiceIntegrationTest extends AbstractLlmServiceIntegrationTest{
+public class LlmServiceIntegrationTest extends AbstractLlmServiceIntegrationTest {
 
     @Autowired
     private LlmService llmService;
 
+
     @Test
-    public void testLlm(){
+    public void testLlm() {
-        llmService.execute(List.of("Wie backe ich eine tiefkühl Pizza?"));
+
+        assumeTrue(true);
     }
 
 }