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

Introduce EventPublisher and EventController for server-sent events (SSE) to enable real-time chat updates.

Browse Source
This commit is contained in:
akastijn 2026-07-18 23:30:49 +02:00
parent 646bbd4145
commit dde66bd608
8 changed files with 243 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatController.java
View File

@ -21,7 +21,8 @@ public class ChatController implements ChatApi {
@Override
public ResponseEntity<Void> sendChatMessages(List<ChatMessageDto> chatMessageDtoList) {
chatMessageDtoList.stream().map(ChatMessageMapper::fromDto).forEach(chatService::addChatMessage);
List<ChatMessage> list = chatMessageDtoList.stream().map(ChatMessageMapper::fromDto).toList();
chatService.addChatMessage(list);
return new ResponseEntity<>(HttpStatus.ACCEPTED);
}

2
backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessage.java
View File

@ -16,5 +16,5 @@ public class ChatMessage {
//TODO [Stijn] [2026-07-18]: Handle channel types
//private final String channel;
private final String messageJson;
private final boolean blocked;
private final boolean notBlocked;
}

6
backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessageMapper.java
View File

@ -16,7 +16,7 @@ public class ChatMessageMapper {
.timestamp(dto.getTimestamp().toInstant())//TODO [Stijn] [2026-07-18]: Check this works, might be sending instant instead
.server(dto.getServer())
.messageJson(dto.getMessage())
.blocked(dto.getBlocked())
.notBlocked(!dto.getBlocked())
.build();
}
@ -26,8 +26,10 @@ public class ChatMessageMapper {
.timestamp(Instant.ofEpochMilli(dao.getTimeStamp()))
.server(dao.getServer())
.messageJson(dao.getMiniMessage())
.blocked(dao.isBlocked())
.notBlocked(!dao.isBlocked())
.build();
}
public static String toJson(ChatMessage chatMessage) {
}
}

67
backend/src/main/java/com/alttd/altitudeweb/controllers/event/EventController.java Normal file
View File

@ -0,0 +1,67 @@
package com.alttd.altitudeweb.controllers.event;
import com.alttd.altitudeweb.controllers.chat.ChatMessageMapper;
import com.alttd.altitudeweb.model.PermissionClaimDto;
import com.alttd.altitudeweb.services.chat.ChatService;
import com.alttd.altitudeweb.services.chat.event_publisher.EventPublisher;
import com.alttd.altitudeweb.services.chat.event_publisher.EventUser;
import jakarta.servlet.http.HttpServletResponse;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.security.core.annotation.AuthenticationPrincipal;
import org.springframework.security.oauth2.jwt.Jwt;
import org.springframework.web.bind.annotation.GetMapping;
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;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
@RequiredArgsConstructor
@Slf4j
@RestController
@RequestMapping("/api/chat/read")
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,
HttpServletResponse response) {
if (jwt == null) {
throw new ResponseStatusException(HttpStatus.UNAUTHORIZED);
}
List<String> authorities = jwt.getClaimAsStringList("authorities");
if (authorities == null || !authorities.contains(PermissionClaimDto.HEAD_MOD.getValue())) {
throw new ResponseStatusException(HttpStatus.FORBIDDEN);
}
response.setHeader("X-Accel-Buffering", "no"); // disables nginx buffering if present
response.setHeader("Cache-Control", "no-cache");
UUID subject;
try {
subject = UUID.fromString(jwt.getSubject());
} catch (IllegalArgumentException e) {
log.error("Invalid UUID in JWT subject: {}", jwt.getSubject());
throw new ResponseStatusException(HttpStatus.BAD_REQUEST);
}
String json = chatService.getMessagesSince(Instant.now().minus(MESSAGE_HISTORY_DURATION))
.stream()
.map(ChatMessageMapper::toJson)
.collect(Collectors.joining(",", "[", "]"));
return eventPublisher.subscribe(new EventUser(subject, authorities), json);
}
}

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

@ -3,10 +3,11 @@ package com.alttd.altitudeweb.services.chat;
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.ChatLogDao;
import com.alttd.altitudeweb.database.chat.ChatLogMapper;
import com.alttd.altitudeweb.database.litebans.HistoryCountMapper;
import com.alttd.altitudeweb.model.PermissionClaimDto;
import com.alttd.altitudeweb.services.chat.event_publisher.EventPublisher;
import com.alttd.altitudeweb.setup.Connection;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
@ -16,13 +17,16 @@ import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.stream.Collectors;
@Slf4j
@Service
@RequiredArgsConstructor
public class ChatService {
private static final Duration MAX_AGE = Duration.ofHours(1);
private final NavigableMap<Instant, ChatMessage> chatMessages = new TreeMap<>();
private final EventPublisher eventPublisher;
@Scheduled(cron = "0 * * * * *")
public void clearOldMessages() {
@ -44,19 +48,39 @@ public class ChatService {
sqlSession.getMapper(ChatLogMapper.class)
.getChatLogs(Instant.now().minus(Duration.ofHours(1)).toEpochMilli())
.stream().map(ChatMessageMapper::fromDao)
.forEach(this::addChatMessage);
.forEach(this::putChatMessage);
} catch (Exception e) {
log.error("Failed to load history count", e);
}
});
}
public void addChatMessage(ChatMessage chatMessage) {
//TODO [Stijn] [2026-07-18]: remove log info after verifying it works
log.info("Adding chat message: {}", chatMessage);
private void putChatMessage(ChatMessage chatMessage) {
chatMessages.put(chatMessage.getTimestamp(), chatMessage);
}
public void addChatMessage(List<ChatMessage> chatMessageList) {
chatMessageList.forEach(this::putChatMessage);
sendMessagesToListeners(chatMessageList);
}
private void sendMessagesToListeners(List<ChatMessage> chatMessageList) {
String jsonMessageListIncludingBlocked = chatMessageList.stream()
.map(ChatMessageMapper::toJson)
.collect(Collectors.joining(",", "[", "]"));
String jsonMessageList = chatMessageList.stream()
.filter(ChatMessage::isNotBlocked)
.map(ChatMessageMapper::toJson)
.collect(Collectors.joining(",", "[", "]"));
eventPublisher.sendToUsers("chat", (eventUser) -> {
if (eventUser.hasPermission(PermissionClaimDto.HEAD_MOD)) {
return jsonMessageListIncludingBlocked;
}
return jsonMessageList;
});
}
public List<ChatMessage> getMessagesSince(Instant instant) {
return chatMessages.tailMap(instant, false).values().stream().toList();
}

103
backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventPublisher.java Normal file
View File

@ -0,0 +1,103 @@
package com.alttd.altitudeweb.services.chat.event_publisher;
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.io.IOException;
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;
@RequiredArgsConstructor
@Service
@Slf4j
public class EventPublisher {
private final Map<UUID, EventUser> eventUserMap = new ConcurrentHashMap<>();
private final Map<UUID, List<SseEmitter>> emitterMap = new ConcurrentHashMap<>();
public SseEmitter subscribe(EventUser eventUser, String initialData) {
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);
log.info("User {} subscribed, active emitters: {}", eventUser.uuid(), countActive());
Runnable cleanup = () -> {
List<SseEmitter> userEmitters = emitterMap.get(eventUser.uuid());
if (userEmitters != null) {
userEmitters.remove(emitter);
if (userEmitters.isEmpty()) {
emitterMap.remove(eventUser.uuid());
eventUserMap.remove(eventUser.uuid());
}
}
log.info("Cleaned up emitter for {}, active emitters: {}", eventUser.uuid(), countActive());
};
// covers: clean disconnect (tab closed normally), server-side timeout, write failure
emitter.onCompletion(cleanup);
emitter.onTimeout(cleanup);
emitter.onError(e -> cleanup.run());
try {
emitter.send(SseEmitter.event().name("connected").data(initialData));
} catch (IOException e) {
cleanup.run();
}
return emitter;
}
// runs every 20s: keeps idle timeouts from firing,
// AND prunes emitters whose sockets are actually dead
@Scheduled(fixedRate = 20_000)
public void heartbeat() {
emitterMap.values().forEach(list -> {
List<SseEmitter> dead = new ArrayList<>();
for (SseEmitter emitter : list) {
try {
emitter.send(SseEmitter.event().comment("heartbeat"));
} catch (IOException e) {
dead.add(emitter);
}
}
list.removeAll(dead);
});
emitterMap.entrySet().removeIf(e -> e.getValue().isEmpty());
}
public void sendToUser(UUID uuid, String eventName, String json) {
List<SseEmitter> userEmitters = emitterMap.get(uuid);
if (userEmitters == null) {
return;
}
List<SseEmitter> dead = new ArrayList<>();
for (SseEmitter emitter : userEmitters) {
try {
emitter.send(SseEmitter.event().name(eventName).data(json));
} catch (IOException e) {
dead.add(emitter);
}
}
userEmitters.removeAll(dead);
}
private long countActive() {
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);
});
}
}

28
backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventUser.java Normal file
View File

@ -0,0 +1,28 @@
package com.alttd.altitudeweb.services.chat.event_publisher;
import com.alttd.altitudeweb.model.PermissionClaimDto;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
public record EventUser(UUID uuid, List<String> authorities) {
public boolean hasPermission(PermissionClaimDto permission) {
return authorities.contains(permission.getValue());
}
@Override
public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) {
return false;
}
EventUser eventUser = (EventUser) o;
return Objects.equals(uuid, eventUser.uuid);
}
@Override
public int hashCode() {
return Objects.hash(uuid);
}
}

8
backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/MessageForUser.java Normal file
View File

@ -0,0 +1,8 @@
package com.alttd.altitudeweb.services.chat.event_publisher;
@FunctionalInterface
public interface MessageForUser {
String get(EventUser eventUser);
}