From b71a2f19b818566239e0c54e0e5ef501b3eb1479 Mon Sep 17 00:00:00 2001 From: akastijn Date: Sun, 2 Aug 2026 17:06:53 +0200 Subject: [PATCH] Refactor EventPublisher and ChatService: replace UUID-based keys with generic unique string keys, add session cleanup callbacks, and update related logic for better extensibility and reusability. --- .../event/CommandsToServerController.java | 52 +++++++++++ .../controllers/event/EventController.java | 6 +- .../services/chat/ChatService.java | 58 ++++++++++++- .../chat/event_publisher/EventPublisher.java | 86 +++++++++---------- .../chat/to_server/ServerMessageService.java | 65 ++++++++++++++ .../chat/to_server/data/ChatFromWeb.java | 16 ++++ .../src/main/resources/application.properties | 1 + .../services/chat/ChatServiceTest.java | 15 ++-- 8 files changed, 239 insertions(+), 60 deletions(-) create mode 100644 backend/src/main/java/com/alttd/altitudeweb/controllers/event/CommandsToServerController.java create mode 100644 backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/ServerMessageService.java create mode 100644 backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/data/ChatFromWeb.java diff --git a/backend/src/main/java/com/alttd/altitudeweb/controllers/event/CommandsToServerController.java b/backend/src/main/java/com/alttd/altitudeweb/controllers/event/CommandsToServerController.java new file mode 100644 index 0000000..0e17870 --- /dev/null +++ b/backend/src/main/java/com/alttd/altitudeweb/controllers/event/CommandsToServerController.java @@ -0,0 +1,52 @@ +package com.alttd.altitudeweb.controllers.event; + +import com.alttd.altitudeweb.services.chat.to_server.ServerMessageService; +import jakarta.servlet.http.HttpServletResponse; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.security.core.token.Token; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.server.ResponseStatusException; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.time.Duration; +import java.time.Instant; + +@RequiredArgsConstructor +@Slf4j +@RestController +@RequestMapping("/api/chat/send") +public class CommandsToServerController { + + private final ServerMessageService serverMessageService; + @Value("${altitudeweb.token}") + private String validToken; + + @GetMapping(path = "/subscribe/{server}", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter subscribe(@AuthenticationPrincipal Token token, + HttpServletResponse response, @PathVariable String server) { + if (validToken == null || validToken.equals("invalid-token")) { + log.error("Invalid token in config"); + throw new ResponseStatusException(HttpStatus.FORBIDDEN); + } + if (token == null) { + throw new ResponseStatusException(HttpStatus.UNAUTHORIZED); + } + + if (!token.getKey().equals(validToken)) { + throw new ResponseStatusException(HttpStatus.FORBIDDEN); + } + + response.setHeader("X-Accel-Buffering", "no"); // disables nginx buffering if present + response.setHeader("Cache-Control", "no-cache"); + + return serverMessageService.subscribe(server); + } +} diff --git a/backend/src/main/java/com/alttd/altitudeweb/controllers/event/EventController.java b/backend/src/main/java/com/alttd/altitudeweb/controllers/event/EventController.java index 4976882..bf09cf5 100644 --- a/backend/src/main/java/com/alttd/altitudeweb/controllers/event/EventController.java +++ b/backend/src/main/java/com/alttd/altitudeweb/controllers/event/EventController.java @@ -3,6 +3,8 @@ package com.alttd.altitudeweb.controllers.event; import com.alttd.altitudeweb.controllers.chat.ChatMessageMapper; import com.alttd.altitudeweb.database.Databases; import com.alttd.altitudeweb.database.chat.ChatLogMapper; +import com.alttd.altitudeweb.database.web_db.chat_session.ChatSession; +import com.alttd.altitudeweb.database.web_db.chat_session.ChatSessionMapper; import com.alttd.altitudeweb.model.PermissionClaimDto; import com.alttd.altitudeweb.services.chat.ChatService; import com.alttd.altitudeweb.services.chat.event_publisher.EventPublisher; @@ -37,7 +39,6 @@ public class EventController { private final static Duration MESSAGE_HISTORY_DURATION = Duration.ofHours(1); private final ChatService chatService; - private final EventPublisher eventPublisher; @GetMapping(path = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter subscribe(@AuthenticationPrincipal Jwt jwt, @@ -68,7 +69,8 @@ public class EventController { .collect(Collectors.joining(",", "[", "]")); Integer partyId = getPartyId(subject); - return eventPublisher.subscribe(new EventUser(subject, authorities, partyId), json); + EventUser eventUser = new EventUser(subject, authorities, partyId); + return chatService.subscribe(eventUser, json); } private Integer getPartyId(UUID subject) { diff --git a/backend/src/main/java/com/alttd/altitudeweb/services/chat/ChatService.java b/backend/src/main/java/com/alttd/altitudeweb/services/chat/ChatService.java index cc0031f..e7cf06e 100644 --- a/backend/src/main/java/com/alttd/altitudeweb/services/chat/ChatService.java +++ b/backend/src/main/java/com/alttd/altitudeweb/services/chat/ChatService.java @@ -4,9 +4,12 @@ import com.alttd.altitudeweb.controllers.chat.ChatMessage; import com.alttd.altitudeweb.controllers.chat.ChatMessageMapper; import com.alttd.altitudeweb.database.Databases; import com.alttd.altitudeweb.database.chat.ChatLogMapper; +import com.alttd.altitudeweb.database.web_db.chat_session.ChatSession; +import com.alttd.altitudeweb.database.web_db.chat_session.ChatSessionMapper; import com.alttd.altitudeweb.model.PermissionClaimDto; import com.alttd.altitudeweb.services.chat.event_publisher.EventPublisher; import com.alttd.altitudeweb.services.chat.event_publisher.EventUser; +import com.alttd.altitudeweb.services.chat.event_publisher.MessageForUser; import com.alttd.altitudeweb.setup.Connection; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -15,6 +18,7 @@ import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.context.event.EventListener; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.time.Duration; import java.time.Instant; @@ -27,10 +31,12 @@ import java.util.stream.Collectors; public class ChatService { private static final Duration MAX_AGE = Duration.ofHours(1); + private final EventPublisher eventPublisher; + private final NavigableMap chatMessages = new TreeMap<>(); + private final Map eventUserMap = new HashMap<>(); + @Value("${chat.allowed-servers}") private String[] allowedServers; - private final NavigableMap chatMessages = new TreeMap<>(); - private final EventPublisher eventPublisher; private long lastMillis = 0; @Scheduled(cron = "0 * * * * *") @@ -61,6 +67,39 @@ public class ChatService { }); } + public SseEmitter subscribe(EventUser eventUser, String json) { + String key = eventUser.uuid().toString(); + eventUserMap.put(key, eventUser); + return eventPublisher.subscribe(key, json, this::handleSessionEnd); + } + + private void handleSessionEnd(String key, Instant sessionStart, Instant sessionEnd) { + EventUser eventUser = eventUserMap.get(key); + if (eventUser == null) { + log.error("Failed to find event user for key {}", key); + return; + } + ChatSession chatSession = ChatSession.builder() + .uuid(eventUser.uuid()) + .session_start(sessionStart) + .session_end(sessionEnd) + .build(); + saveSession(chatSession); + } + + private void saveSession(ChatSession chatSession) { + Connection.getConnection(Databases.DEFAULT) + .runQuery(sqlSession -> { + log.debug("Saving chat session"); + try { + sqlSession.getMapper(ChatSessionMapper.class).storeSession(chatSession); + log.debug("Saved chat session"); + } catch (Exception e) { + log.error("Failed to save chat session", e); + } + }); + } + private synchronized void setUniqueTimestamp(ChatMessage chatMessage) { long timestamp = chatMessage.getTimestamp().toEpochMilli(); if (timestamp == lastMillis) { @@ -88,7 +127,7 @@ public class ChatService { jsonCache.put(message, ChatMessageMapper.toJson(message)); } - eventPublisher.sendToUsers("chat", (eventUser) -> + sendToUsers("chat", (eventUser) -> chatMessageList.stream() .filter(chatMessage -> shouldReceive(eventUser, chatMessage)) .map(jsonCache::get) @@ -96,6 +135,19 @@ public class ChatService { ); } + + /** + * Broadcasts to every currently subscribed key. The message content for + * each key is resolved via messageForUser, which now works off the + * unique key itself rather than an EventUser instance. + */ + public void sendToUsers(String eventName, MessageForUser messageForUser) { + eventUserMap.forEach((key, eventUser) -> { + String json = messageForUser.get(eventUser); + eventPublisher.sendToUser(key, eventName, json); + }); + } + private boolean shouldReceive(EventUser eventUser, ChatMessage chatMessage) { if (eventUser.hasPermission(PermissionClaimDto.HEAD_MOD)) { return true; diff --git a/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventPublisher.java b/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventPublisher.java index c0d3417..fd9c8ec 100644 --- a/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventPublisher.java +++ b/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventPublisher.java @@ -1,10 +1,5 @@ package com.alttd.altitudeweb.services.chat.event_publisher; -import com.alttd.altitudeweb.database.Databases; -import com.alttd.altitudeweb.database.web_db.KeyPairMapper; -import com.alttd.altitudeweb.database.web_db.chat_session.ChatSession; -import com.alttd.altitudeweb.database.web_db.chat_session.ChatSessionMapper; -import com.alttd.altitudeweb.setup.Connection; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; @@ -17,27 +12,44 @@ import java.time.Instant; import java.util.ArrayList; import java.util.List; import java.util.Map; -import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +/** + * Generic SSE event publisher, keyed by an arbitrary "unique" string + * The same key may be used by several concurrent connections + * (e.g., a user/server with multiple open tabs or connections). + * Each subscribe() call adds its own emitter under that key + * without evicting the others. + */ @RequiredArgsConstructor @Service @Slf4j public class EventPublisher { - private final Map eventUserMap = new ConcurrentHashMap<>(); - private final Map> emitterMap = new ConcurrentHashMap<>(); + private final Map> emitterMap = new ConcurrentHashMap<>(); - public SseEmitter subscribe(EventUser eventUser, String initialData) { + /** + * Invoked once a given emitter for uniqueKey is cleaned up (completed, + * timed out, or errored). Callers use this to do whatever bookkeeping + * they need (e.g., persisting a Session) using sessionStart/sessionEnd. + */ + @FunctionalInterface + public interface SessionCleanupCallback { + void onCleanup(String uniqueKey, Instant sessionStart, Instant sessionEnd); + } + public SseEmitter subscribe(String uniqueKey, SessionCleanupCallback cleanupCallback) { + return subscribe(uniqueKey, null, cleanupCallback); + } + + public SseEmitter subscribe(String uniqueKey, String initialData, SessionCleanupCallback cleanupCallback) { SseEmitter emitter = new SseEmitter(0L); // no built-in timeout; heartbeat governs liveness - emitterMap.computeIfAbsent(eventUser.uuid(), k -> new CopyOnWriteArrayList<>()).add(emitter); - eventUserMap.put(eventUser.uuid(), eventUser); + emitterMap.computeIfAbsent(uniqueKey, k -> new CopyOnWriteArrayList<>()).add(emitter); Instant sessionStart = Instant.now(); - log.info("User {} subscribed, active emitters: {}", eventUser.uuid(), countActive()); + log.info("Key {} subscribed, active emitters: {}", uniqueKey, countActive()); - Runnable cleanup = () -> handleCleanup(eventUser, emitter, sessionStart); + Runnable cleanup = () -> handleCleanup(uniqueKey, emitter, sessionStart, cleanupCallback); // covers: clean disconnect (tab closed normally), server-side timeout, write failure emitter.onCompletion(cleanup); @@ -45,7 +57,11 @@ public class EventPublisher { emitter.onError(e -> cleanup.run()); try { - emitter.send(SseEmitter.event().name("connected").data(initialData)); + if (initialData != null) { + emitter.send(SseEmitter.event().name("connected").data(initialData)); + } else { + emitter.send(SseEmitter.event().name("connected")); + } } catch (IOException e) { cleanup.run(); } @@ -71,8 +87,8 @@ public class EventPublisher { emitterMap.entrySet().removeIf(e -> e.getValue().isEmpty()); } - public void sendToUser(UUID uuid, String eventName, String json) { - List userEmitters = emitterMap.get(uuid); + public void sendToUser(String uniqueKey, String eventName, String json) { + List userEmitters = emitterMap.get(uniqueKey); if (userEmitters == null) { return; } @@ -92,45 +108,23 @@ public class EventPublisher { return emitterMap.values().stream().mapToLong(List::size).sum(); } - public void sendToUsers(String eventName, MessageForUser messagesForUser) { - eventUserMap.values().forEach(emitter -> { - String json = messagesForUser.get(emitter); - sendToUser(emitter.uuid(), eventName, json); - }); - } - - private void handleCleanup(EventUser eventUser, SseEmitter emitter, Instant sessionStart) { - List userEmitters = emitterMap.get(eventUser.uuid()); + private void handleCleanup(String uniqueKey, SseEmitter emitter, Instant sessionStart, + SessionCleanupCallback cleanupCallback) { + List userEmitters = emitterMap.get(uniqueKey); if (userEmitters != null) { userEmitters.remove(emitter); if (userEmitters.isEmpty()) { - emitterMap.remove(eventUser.uuid()); - eventUserMap.remove(eventUser.uuid()); + emitterMap.remove(uniqueKey); } } Instant sessionEnd = Instant.now(); Duration between = Duration.between(sessionStart, sessionEnd); log.info("Cleaned up emitter for {}, active emitters: {}, session duration: {}", - eventUser.uuid(), countActive(), between + uniqueKey, countActive(), between ); - ChatSession chatSession = ChatSession.builder() - .uuid(eventUser.uuid()) - .session_start(sessionStart) - .session_end(sessionEnd) - .build(); - saveSession(chatSession); - } - private void saveSession(ChatSession chatSession) { - Connection.getConnection(Databases.DEFAULT) - .runQuery(sqlSession -> { - log.debug("Saving chat session"); - try { - sqlSession.getMapper(ChatSessionMapper.class).storeSession(chatSession); - log.debug("Saved chat session"); - } catch (Exception e) { - log.error("Failed to save chat session", e); - } - }); + if (cleanupCallback != null) { + cleanupCallback.onCleanup(uniqueKey, sessionStart, sessionEnd); + } } } diff --git a/backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/ServerMessageService.java b/backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/ServerMessageService.java new file mode 100644 index 0000000..db534ba --- /dev/null +++ b/backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/ServerMessageService.java @@ -0,0 +1,65 @@ +package com.alttd.altitudeweb.services.chat.to_server; + +import com.alttd.altitudeweb.services.chat.event_publisher.EventPublisher; +import com.alttd.altitudeweb.services.chat.event_publisher.EventUser; +import com.alttd.altitudeweb.services.chat.to_server.data.ChatFromWeb; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashSet; +import java.util.Set; +import java.util.UUID; + +@Slf4j +@Service +@RequiredArgsConstructor +public class ServerMessageService { + + private final Set servers = new HashSet<>(); + private final EventPublisher eventPublisher; + + public SseEmitter subscribe(String server) { + servers.add(server); + return eventPublisher.subscribe(server, this::handleSessionEnd); + } + + private void handleSessionEnd(String key, Instant sessionStart, Instant sessionEnd) { + log.info("Server session ended for {}. Active from {} for {}", key, sessionStart, Duration.between(sessionStart, sessionEnd)); + } + + public boolean sendMessage(String server, String channel, String json) { + if (!servers.contains(server)) { + log.warn("Server {} is not connected, not sending message", server); + return false; + } + eventPublisher.sendToUser(server, channel, json); + return true; + } + + @Scheduled(fixedRate = 10_000) + public void sendTestMessage() { + ChatFromWeb chatFromWeb = ChatFromWeb.builder().sender(UUID.fromString("55e46bc3-2a29-4c53-850f-dbd944dc5c5f")) + .message("This is a test a cool test").build(); + if (!sendMessage("alpha", "web_chat", toJson(chatFromWeb))) { + log.warn("Failed to send test message"); + } + log.info("Sent test message"); + } + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + public static String toJson(Object object) { + try { + return OBJECT_MAPPER.writeValueAsString(object); + } catch (JsonProcessingException e) { + throw new IllegalStateException("Failed to serialize object to JSON", e); + } + } +} diff --git a/backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/data/ChatFromWeb.java b/backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/data/ChatFromWeb.java new file mode 100644 index 0000000..afd10e5 --- /dev/null +++ b/backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/data/ChatFromWeb.java @@ -0,0 +1,16 @@ +package com.alttd.altitudeweb.services.chat.to_server.data; + +import lombok.Builder; +import lombok.Data; +import lombok.Getter; + +import java.util.UUID; + +@Builder +@Getter +public class ChatFromWeb { + + private final UUID sender; + private final String message; + +} diff --git a/backend/src/main/resources/application.properties b/backend/src/main/resources/application.properties index 4b2f475..3d01bcc 100644 --- a/backend/src/main/resources/application.properties +++ b/backend/src/main/resources/application.properties @@ -24,3 +24,4 @@ management.endpoints.web.exposure.include=mappings server.forward-headers-strategy=native server.tomcat.remoteip.remote-ip-header=CF-Connecting-IP server.tomcat.remoteip.trusted-proxies=10\\.0\\.0\\.103 +altitudeweb.token=${TOKEN:invalid-token} diff --git a/backend/src/test/java/com/alttd/altitudeweb/services/chat/ChatServiceTest.java b/backend/src/test/java/com/alttd/altitudeweb/services/chat/ChatServiceTest.java index cc69d54..91763b2 100644 --- a/backend/src/test/java/com/alttd/altitudeweb/services/chat/ChatServiceTest.java +++ b/backend/src/test/java/com/alttd/altitudeweb/services/chat/ChatServiceTest.java @@ -3,7 +3,6 @@ package com.alttd.altitudeweb.services.chat; import com.alttd.altitudeweb.controllers.chat.ChatMessage; import com.alttd.altitudeweb.controllers.chat.ChatMessageType; import com.alttd.altitudeweb.model.PermissionClaimDto; -import com.alttd.altitudeweb.services.chat.event_publisher.EventPublisher; import com.alttd.altitudeweb.services.chat.event_publisher.EventUser; import com.alttd.altitudeweb.services.chat.event_publisher.MessageForUser; import org.junit.jupiter.api.BeforeEach; @@ -23,12 +22,10 @@ import static org.mockito.Mockito.*; class ChatServiceTest { private ChatService chatService; - private EventPublisher eventPublisher; @BeforeEach void setUp() { - eventPublisher = mock(EventPublisher.class); - chatService = new ChatService(eventPublisher); + chatService = spy(new ChatService()); ReflectionTestUtils.setField(chatService, "allowedServers", new String[]{"server1"}); } @@ -46,7 +43,7 @@ class ChatServiceTest { chatService.addChatMessage(List.of(message)); ArgumentCaptor captor = ArgumentCaptor.forClass(MessageForUser.class); - verify(eventPublisher).sendToUsers(anyString(), captor.capture()); + verify(chatService).sendToUsers(anyString(), captor.capture()); MessageForUser messageForUser = captor.getValue(); EventUser user = new EventUser(UUID.randomUUID(), List.of(), null); @@ -69,7 +66,7 @@ class ChatServiceTest { chatService.addChatMessage(List.of(message)); ArgumentCaptor captor = ArgumentCaptor.forClass(MessageForUser.class); - verify(eventPublisher).sendToUsers(anyString(), captor.capture()); + verify(chatService).sendToUsers(anyString(), captor.capture()); MessageForUser messageForUser = captor.getValue(); @@ -97,7 +94,7 @@ class ChatServiceTest { chatService.addChatMessage(List.of(message)); ArgumentCaptor captor = ArgumentCaptor.forClass(MessageForUser.class); - verify(eventPublisher).sendToUsers(anyString(), captor.capture()); + verify(chatService).sendToUsers(anyString(), captor.capture()); MessageForUser messageForUser = captor.getValue(); @@ -123,7 +120,7 @@ class ChatServiceTest { chatService.addChatMessage(List.of(message)); ArgumentCaptor captor = ArgumentCaptor.forClass(MessageForUser.class); - verify(eventPublisher).sendToUsers(anyString(), captor.capture()); + verify(chatService).sendToUsers(anyString(), captor.capture()); MessageForUser messageForUser = captor.getValue(); EventUser user = new EventUser(UUID.randomUUID(), List.of(), null); @@ -144,7 +141,7 @@ class ChatServiceTest { chatService.addChatMessage(List.of(message)); ArgumentCaptor captor = ArgumentCaptor.forClass(MessageForUser.class); - verify(eventPublisher).sendToUsers(anyString(), captor.capture()); + verify(chatService).sendToUsers(anyString(), captor.capture()); MessageForUser messageForUser = captor.getValue(); EventUser regularUser = new EventUser(UUID.randomUUID(), List.of(), null);