diff --git a/api/build.gradle.kts b/api/build.gradle.kts index 3e37815..0a46876 100644 --- a/api/build.gradle.kts +++ b/api/build.gradle.kts @@ -16,6 +16,7 @@ dependencies { //API validation implementation("org.hibernate.validator:hibernate-validator:9.0.1.Final") implementation("org.glassfish:jakarta.el:5.0.0-M1") + implementation("com.fasterxml.jackson.core:jackson-databind:2.22.1") } publishing { diff --git a/api/src/main/java/com/alttd/chat/config/Config.java b/api/src/main/java/com/alttd/chat/config/Config.java index 84ca66d..4ca394c 100755 --- a/api/src/main/java/com/alttd/chat/config/Config.java +++ b/api/src/main/java/com/alttd/chat/config/Config.java @@ -652,8 +652,12 @@ public final class Config { } public static String CHAT_WEB_SERVER_BASE_URL = "http://10.0.0.121:8080"; + public static String CHAT_WEB_REGISTER_TO_BASE_URL = "https://alttd.com"; + public static String CHAT_WEB_TOKEN = "invalid-token"; private static void webServerSettings() { CHAT_WEB_SERVER_BASE_URL = getString("web-server.base-url", CHAT_WEB_SERVER_BASE_URL); + CHAT_WEB_REGISTER_TO_BASE_URL = getString("web-server.register-to-base-url", CHAT_WEB_REGISTER_TO_BASE_URL); + CHAT_WEB_TOKEN = getString("web-server.token", CHAT_WEB_TOKEN); } } diff --git a/api/src/main/java/com/alttd/chat/web/SseSubscribeClient.java b/api/src/main/java/com/alttd/chat/web/SseSubscribeClient.java new file mode 100644 index 0000000..96f6660 --- /dev/null +++ b/api/src/main/java/com/alttd/chat/web/SseSubscribeClient.java @@ -0,0 +1,194 @@ +package com.alttd.chat.web; + +import com.alttd.chat.config.Config; +import com.alttd.chat.util.ALogger; +import com.fasterxml.jackson.databind.ObjectMapper; + +import java.io.BufferedReader; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Client for subscribing to: GET /api/chat/send/subscribe/{server} served as text/event-stream (SseEmitter) by + * CommandsToServerController. + *

+ * Handlers for individual SSE event types can be registered up front via {@link #register(String, WebHandler)}. Each + * handler says "when an event named X arrives, deserialize its JSON payload as type T and pass it to this handler." + * Unregistered event names are logged and skipped, not thrown. + *

+ * If the connection fails or drops for any reason, the client logs it and retries once a minute, indefinitely, until + * stop() is called or the thread is interrupted. + *

+ * Server-side auth: the controller checks @AuthenticationPrincipal Token token, then token.getKey().equals(validToken). + * That principal is populated from the "Authorization: Bearer " header. validToken is currently hardcoded + * server-side; swap TOKEN below for a config value once that TODO is resolved. + */ +public class SseSubscribeClient implements Runnable { + + private static final Logger log = Logger.getLogger(SseSubscribeClient.class.getName()); + private static final Duration RETRY_DELAY = Duration.ofMinutes(1); + + private final String baseUrl; + private final String server; + private final HttpClient httpClient; + private final ObjectMapper objectMapper = new ObjectMapper(); + private final Map> handlers = new ConcurrentHashMap<>(); + + private volatile boolean running = false; + + public SseSubscribeClient(String baseUrl, String server, String token) { + this.baseUrl = baseUrl; + this.server = server; + this.httpClient = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(10)) + .build(); + } + + /** + * Register a handler for a given SSE event name. When an event with this name arrives, its data payload is + * deserialized using the handler's type and passed to the handler. + */ + public void register(String eventName, WebHandler handler) { + handlers.put(eventName, handler); + } + + /** + * Runs the subscribe/retry loop on the calling thread. Blocks until stop() is called or the thread is interrupted. + */ + @Override + public void run() { + running = true; + while (running && !Thread.currentThread().isInterrupted()) { + String token = Config.CHAT_WEB_TOKEN; + if (token.equals("invalid-token")) { + log.warning("Invalid token provided for SSE connection as '" + server + "'"); + sleepBeforeRetry(); + continue; + } + try { + connectAndListen(token); + log.warning("SSE stream as '" + server + "' ended; reconnecting in " + RETRY_DELAY.toMinutes() + " minute(s)"); + } catch (Exception e) { + log.log(Level.WARNING, "SSE connection to '" + server + "' failed; retrying in " + + RETRY_DELAY.toMinutes() + " minute(s)", e + ); + } + + if (running) { + sleepBeforeRetry(); + } + } + } + + public void stop() { + running = false; + } + + private void connectAndListen(String token) throws Exception { + HttpResponse response = openConnection(token); + checkStatus(response); + readEvents(response.body()); + } + + private HttpResponse openConnection(String token) throws Exception { + HttpRequest request = HttpRequest.newBuilder() + .uri(URI.create(baseUrl + "/api/chat/server/send/subscribe/" + server)) + .header("X-Altitude-Token", token) + .header("Accept", "text/event-stream") + .GET() + .build(); + + return httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream()); + } + + private void checkStatus(HttpResponse response) { + int status = response.statusCode(); + if (status == 401) { + throw new RuntimeException("Unauthorized - missing/invalid token"); + } + if (status == 403) { + throw new RuntimeException("Forbidden - token did not match server's validToken"); + } + if (status != 200) { + throw new RuntimeException("Unexpected status: " + status); + } + ALogger.info("SSE connection as '" + server + "' established"); + } + + /** + * Reads the SSE stream line by line, accumulating one event at a time, and dispatches each complete event as it's + * parsed. + */ + private void readEvents(InputStream body) throws Exception { + try (BufferedReader reader = new BufferedReader(new InputStreamReader(body, StandardCharsets.UTF_8))) { + String eventName = null; + StringBuilder dataBuffer = new StringBuilder(); + String line; + + while ((line = reader.readLine()) != null) { + if (line.isEmpty()) { + if (!dataBuffer.isEmpty()) { + dispatch(eventName, dataBuffer.toString()); + } + eventName = null; + dataBuffer.setLength(0); + continue; + } + + if (line.startsWith("event:")) { + eventName = line.substring(6).trim(); + } else if (line.startsWith("data:")) { + if (!dataBuffer.isEmpty()) { + dataBuffer.append('\n'); + } + dataBuffer.append(line.substring(5).trim()); + } + // lines starting with ":" are comments/heartbeats - ignored + } + } + } + + private void dispatch(String eventName, String data) { + if (eventName == null) { + log.warning("Received event with no 'event:' name, ignoring. Data: " + data); + return; + } + + WebHandler handler = handlers.get(eventName); + if (handler == null) { + log.warning("No handler registered for event type '" + eventName + "', ignoring"); + return; + } + + dispatchTyped(eventName, handler, data); + } + + // Captures the wildcard type from the handler so parsing/casting is type-safe. + private void dispatchTyped(String eventName, WebHandler handler, String data) { + try { + T event = objectMapper.readValue(data, handler.type()); + handler.handle(event); + } catch (Exception e) { + log.log(Level.WARNING, "Failed to parse/handle event '" + eventName + "'", e); + } + } + + private void sleepBeforeRetry() { + try { + Thread.sleep(RETRY_DELAY.toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + running = false; + } + } +} diff --git a/api/src/main/java/com/alttd/chat/web/WebHandler.java b/api/src/main/java/com/alttd/chat/web/WebHandler.java new file mode 100644 index 0000000..72ef0fd --- /dev/null +++ b/api/src/main/java/com/alttd/chat/web/WebHandler.java @@ -0,0 +1,9 @@ +package com.alttd.chat.web; + +public interface WebHandler { + + Class type(); + + void handle(T event); + +} diff --git a/api/src/main/java/com/alttd/chat/web/handler_class/ChatFromWeb.java b/api/src/main/java/com/alttd/chat/web/handler_class/ChatFromWeb.java new file mode 100644 index 0000000..6c27052 --- /dev/null +++ b/api/src/main/java/com/alttd/chat/web/handler_class/ChatFromWeb.java @@ -0,0 +1,15 @@ +package com.alttd.chat.web.handler_class; + +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.UUID; + +@Data +@NoArgsConstructor +public class ChatFromWeb { + + private UUID sender; + private String message; + +} diff --git a/galaxy/src/main/java/com/alttd/chat/ChatPlugin.java b/galaxy/src/main/java/com/alttd/chat/ChatPlugin.java index 6b42292..f5b1fdc 100755 --- a/galaxy/src/main/java/com/alttd/chat/ChatPlugin.java +++ b/galaxy/src/main/java/com/alttd/chat/ChatPlugin.java @@ -1,6 +1,7 @@ package com.alttd.chat; import com.alttd.chat.chat_web.ChatMessageSender; +import com.alttd.chat.chat_web.handlers.WebChatHandler; import com.alttd.chat.commands.*; import com.alttd.chat.config.Config; import com.alttd.chat.config.ServerConfig; @@ -15,6 +16,8 @@ import com.alttd.chat.objects.chat_log.ChatLogHandler; import com.alttd.chat.util.ALogger; import com.alttd.chat.util.ServerName; import com.alttd.chat.util.Utility; +import com.alttd.chat.web.SseSubscribeClient; +import lombok.Getter; import org.bukkit.Bukkit; import org.bukkit.command.CommandExecutor; import org.bukkit.event.Listener; @@ -24,12 +27,15 @@ import java.util.Objects; public class ChatPlugin extends JavaPlugin { + @Getter private static ChatPlugin instance; private ChatAPI chatAPI; + @Getter private ChatHandler chatHandler; private ServerConfig serverConfig; + private SseSubscribeClient sseSubscribeClient; @Override public void onEnable() { @@ -78,10 +84,23 @@ public class ChatPlugin extends JavaPlugin { getServer().getMessenger().registerIncomingPluginChannel(this, messageChannel, nicknamesEvents); getServer().getPluginManager().registerEvents(nicknamesEvents, this); registerCommand("nick", new Nicknames()); + + sseSubscribeClient = new SseSubscribeClient( + Config.CHAT_WEB_REGISTER_TO_BASE_URL, + getServer().getServerName(), + Config.CHAT_WEB_TOKEN + ); + new Thread(sseSubscribeClient).start(); + registerWebHandlers(chatMessageSender, sseSubscribeClient); + } + + private void registerWebHandlers(ChatMessageSender chatMessageSender, SseSubscribeClient sseSubscribeClient) { + sseSubscribeClient.register("web_chat", new WebChatHandler(chatMessageSender)); } @Override public void onDisable() { + sseSubscribeClient.stop(); instance = null; } @@ -95,14 +114,6 @@ public class ChatPlugin extends JavaPlugin { Objects.requireNonNull(getCommand(commandName)).setExecutor(commandExecutor); } - public static ChatPlugin getInstance() { - return instance; - } - - public ChatHandler getChatHandler() { - return chatHandler; - } - public boolean serverGlobalChatEnabled() { return serverConfig.GLOBALCHAT; } diff --git a/galaxy/src/main/java/com/alttd/chat/chat_web/handlers/WebChatHandler.java b/galaxy/src/main/java/com/alttd/chat/chat_web/handlers/WebChatHandler.java new file mode 100644 index 0000000..36cb3ef --- /dev/null +++ b/galaxy/src/main/java/com/alttd/chat/chat_web/handlers/WebChatHandler.java @@ -0,0 +1,24 @@ +package com.alttd.chat.chat_web.handlers; + +import com.alttd.chat.chat_web.ChatMessageSender; +import com.alttd.chat.web.WebHandler; +import com.alttd.chat.web.handler_class.ChatFromWeb; + +public class WebChatHandler implements WebHandler { + + private final ChatMessageSender chatMessageSender; + + public WebChatHandler(ChatMessageSender chatMessageSender) { + this.chatMessageSender = chatMessageSender; + } + + @Override + public Class type() { + return ChatFromWeb.class; + } + + @Override + public void handle(ChatFromWeb chatFromWeb) { + this.chatMessageSender.sendMessage(chatFromWeb.getSender(), chatFromWeb.getMessage()); + } +}