Altitude/AltitudeWeb
5
0
Fork 0
Code Issues 12 Pull Requests Actions Packages Projects Releases Wiki Activity

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.

Browse Source
This commit is contained in:
akastijn 2026-08-02 17:06:53 +02:00
parent f7edde65d6
commit b71a2f19b8
8 changed files with 239 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
backend/src/main/java/com/alttd/altitudeweb/controllers/event/CommandsToServerController.java Normal file
View File

@ -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);
}
}

6
backend/src/main/java/com/alttd/altitudeweb/controllers/event/EventController.java
View File

@ -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) {

58
backend/src/main/java/com/alttd/altitudeweb/services/chat/ChatService.java
View File

@ -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<Instant, ChatMessage> chatMessages = new TreeMap<>();
private final Map<String, EventUser> eventUserMap = new HashMap<>();
@Value("${chat.allowed-servers}")
private String[] allowedServers;
private final NavigableMap<Instant, ChatMessage> 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;

84
backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventPublisher.java
View File

@ -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<UUID, EventUser> eventUserMap = new ConcurrentHashMap<>();
private final Map<UUID, List<SseEmitter>> emitterMap = new ConcurrentHashMap<>();
private final Map<String, List<SseEmitter>> 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 {
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<SseEmitter> userEmitters = emitterMap.get(uuid);
public void sendToUser(String uniqueKey, String eventName, String json) {
List<SseEmitter> 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<SseEmitter> userEmitters = emitterMap.get(eventUser.uuid());
private void handleCleanup(String uniqueKey, SseEmitter emitter, Instant sessionStart,
SessionCleanupCallback cleanupCallback) {
List<SseEmitter> 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);
}
}
}

65
backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/ServerMessageService.java Normal file
View File

@ -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<String> 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 <red>a cool test</red>").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);
}
}
}

16
backend/src/main/java/com/alttd/altitudeweb/services/chat/to_server/data/ChatFromWeb.java Normal file
View File

@ -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;
}

1
backend/src/main/resources/application.properties
View File

@ -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}

15
backend/src/test/java/com/alttd/altitudeweb/services/chat/ChatServiceTest.java
View File

@ -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<MessageForUser> 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<MessageForUser> 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<MessageForUser> 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<MessageForUser> 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<MessageForUser> 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);