diff --git a/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatController.java b/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatController.java index e368ebc..f69dab7 100644 --- a/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatController.java +++ b/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatController.java @@ -21,7 +21,8 @@ public class ChatController implements ChatApi { @Override public ResponseEntity sendChatMessages(List chatMessageDtoList) { - chatMessageDtoList.stream().map(ChatMessageMapper::fromDto).forEach(chatService::addChatMessage); + List list = chatMessageDtoList.stream().map(ChatMessageMapper::fromDto).toList(); + chatService.addChatMessage(list); return new ResponseEntity<>(HttpStatus.ACCEPTED); } diff --git a/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessage.java b/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessage.java index b168491..e9efba8 100644 --- a/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessage.java +++ b/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessage.java @@ -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; } diff --git a/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessageMapper.java b/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessageMapper.java index 8d4fb7b..a95c69b 100644 --- a/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessageMapper.java +++ b/backend/src/main/java/com/alttd/altitudeweb/controllers/chat/ChatMessageMapper.java @@ -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) { + } } 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 new file mode 100644 index 0000000..e91dd1a --- /dev/null +++ b/backend/src/main/java/com/alttd/altitudeweb/controllers/event/EventController.java @@ -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 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); + } +} 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 f361d7a..e933ff5 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 @@ -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 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 chatMessageList) { + chatMessageList.forEach(this::putChatMessage); + sendMessagesToListeners(chatMessageList); + } + + private void sendMessagesToListeners(List 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 getMessagesSince(Instant instant) { return chatMessages.tailMap(instant, false).values().stream().toList(); } 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 new file mode 100644 index 0000000..4a0467f --- /dev/null +++ b/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventPublisher.java @@ -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 eventUserMap = new ConcurrentHashMap<>(); + private final Map> 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 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 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 userEmitters = emitterMap.get(uuid); + if (userEmitters == null) { + return; + } + + List 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); + }); + } +} diff --git a/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventUser.java b/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventUser.java new file mode 100644 index 0000000..c713063 --- /dev/null +++ b/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/EventUser.java @@ -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 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); + } +} diff --git a/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/MessageForUser.java b/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/MessageForUser.java new file mode 100644 index 0000000..75fe537 --- /dev/null +++ b/backend/src/main/java/com/alttd/altitudeweb/services/chat/event_publisher/MessageForUser.java @@ -0,0 +1,8 @@ +package com.alttd.altitudeweb.services.chat.event_publisher; + +@FunctionalInterface +public interface MessageForUser { + + String get(EventUser eventUser); + +}