package dev.streamchat; import com.google.gson.JsonObject; import com.google.gson.JsonParser; import net.minecraft.client.Minecraft; import net.minecraft.network.chat.Component; import net.minecraft.client.gui.components.toasts.SystemToast; import java.net.URI; import java.net.http.HttpClient; import java.net.http.WebSocket; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.time.Duration; import java.time.Instant; import java.util.EnumMap; import java.util.Locale; import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletionStage; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; public final class ConnectionManager { private static final Logger LOGGER = Logger.getLogger("StreamChat"); private static final String[] TWITCH_EVENTS = { "ChatMessage", "Follow", "Sub", "ReSub", "GiftSub", "Cheer", "Raid" }; private static final String[] YOUTUBE_EVENTS = { "Message", "SuperChat", "SuperSticker", "MemberMilestone", "NewSponsor", "GiftMembershipReceived" }; private final Minecraft client; private final StreamChatConfig config; private final HttpClient httpClient = HttpClient.newBuilder() .connectTimeout(Duration.ofSeconds(10)) .build(); private final ScheduledExecutorService retryExecutor = Executors.newSingleThreadScheduledExecutor(task -> { Thread thread = new Thread(task, "Stream Chat reconnect"); thread.setDaemon(true); return thread; }); private final Map sockets = new EnumMap<>(Platform.class); private final Map statuses = new EnumMap<>(Platform.class); private final Map lastEvents = new EnumMap<>(Platform.class); private final Map retryCounts = new EnumMap<>(Platform.class); private final java.util.Set scheduledRetries = java.util.EnumSet.noneOf(Platform.class); private boolean manuallyPaused; private volatile long connectionGeneration; private String authenticationRequestId; public ConnectionManager(Minecraft client, StreamChatConfig config) { this.client = client; this.config = config; for (Platform platform : Platform.values()) { statuses.put(platform, "Disconnected"); } } public synchronized void start() { manuallyPaused = false; connectEnabledPlatforms(); } public synchronized void toggleConnections() { manuallyPaused = !manuallyPaused; if (manuallyPaused) { closeAll(); } else { connectEnabledPlatforms(); } } public synchronized void configurationChanged() { if (!manuallyPaused) { closeAll(); connectEnabledPlatforms(); } } public synchronized String status(Platform platform) { return statuses.getOrDefault(platform, "Disconnected"); } public synchronized String lastEvent(Platform platform) { Instant timestamp = lastEvents.get(platform); return timestamp == null ? "Never" : timestamp.toString(); } private void connectEnabledPlatforms() { if (config.streamerBotEnabled && (config.twitchEnabled || config.youtubeEnabled)) { connect(Platform.TWITCH, config.streamerBotUrl); } else { statuses.put(Platform.TWITCH, "Disabled"); statuses.put(Platform.YOUTUBE, "Disabled"); } if (config.tikFinityEnabled && config.tiktokEnabled) { connect(Platform.TIKTOK, config.tikFinityUrl); } else { statuses.put(Platform.TIKTOK, "Disabled"); } } private synchronized void connect(Platform platform, String address) { long generation = connectionGeneration; statuses.put(platform, "Connecting"); try { httpClient.newWebSocketBuilder() .connectTimeout(Duration.ofSeconds(10)) .buildAsync(URI.create(address), new Listener(platform, generation)) .whenComplete((socket, error) -> { synchronized (ConnectionManager.this) { if (generation != connectionGeneration) { if (socket != null) socket.sendClose(WebSocket.NORMAL_CLOSURE, "Configuration changed"); return; } if (error != null) { statuses.put(platform, "Error: " + error.getMessage()); LOGGER.log(Level.WARNING, platform + " connection failed", error); scheduleReconnect(platform, address, generation); } else { sockets.put(platform, socket); } } }); } catch (IllegalArgumentException exception) { statuses.put(platform, "Error: " + exception.getMessage()); LOGGER.log(Level.WARNING, "Invalid WebSocket address for " + platform, exception); } } private synchronized void closeAll() { connectionGeneration++; for (Map.Entry entry : sockets.entrySet()) { entry.getValue().sendClose(WebSocket.NORMAL_CLOSURE, "Disconnected by user"); statuses.put(entry.getKey(), "Disconnected"); if (entry.getKey() == Platform.TWITCH && config.youtubeEnabled) { statuses.put(Platform.YOUTUBE, "Disconnected"); } } sockets.clear(); scheduledRetries.clear(); retryCounts.clear(); } private void sendStreamerBotSubscription(WebSocket socket) { JsonObject events = new JsonObject(); if (config.twitchEnabled) { events.add("Twitch", strings(TWITCH_EVENTS)); } if (config.youtubeEnabled) { events.add("YouTube", strings(YOUTUBE_EVENTS)); } JsonObject request = new JsonObject(); request.addProperty("request", "Subscribe"); request.addProperty("id", UUID.randomUUID().toString()); request.add("events", events); socket.sendText(request.toString(), true); } private com.google.gson.JsonArray strings(String[] values) { com.google.gson.JsonArray array = new com.google.gson.JsonArray(); for (String value : values) array.add(value); return array; } private void receive(Platform socketPlatform, String raw) { try { JsonObject root = JsonParser.parseString(raw).getAsJsonObject(); if (socketPlatform == Platform.TWITCH && handleStreamerBotControl(root)) return; if (config.debugLogging) LOGGER.info("Received from " + socketPlatform + ": " + raw); StreamEvent event = socketPlatform == Platform.TIKTOK ? StreamEventParser.parseTikFinity(raw) : StreamEventParser.parseStreamerBot(raw); if (event.platform() == Platform.TIKTOK && "config".equals(event.type())) return; synchronized (this) { lastEvents.put(event.platform(), Instant.now()); statuses.put(event.platform(), "Connected"); } if (!isKnownEvent(event)) { LOGGER.warning("Unrecognized " + event.platform() + " event: " + raw); } if (!config.isEnabled(event.platform())) return; DisplayTarget target = config.getRoute(event); if (target != DisplayTarget.HIDDEN) { client.execute(() -> display(event, target)); } } catch (RuntimeException exception) { LOGGER.log(Level.WARNING, "Could not parse event from " + socketPlatform + ": " + raw, exception); } } private synchronized void scheduleReconnect(Platform platform, String address, long generation) { if (generation != connectionGeneration || manuallyPaused || !isConnectionEnabled(platform) || !scheduledRetries.add(platform)) { return; } int attempts = retryCounts.merge(platform, 1, Integer::sum); long delaySeconds = Math.min(30, 1L << Math.min(attempts, 5)); statuses.put(platform, "Reconnecting in " + delaySeconds + "s"); retryExecutor.schedule(() -> { synchronized (ConnectionManager.this) { scheduledRetries.remove(platform); if (generation == connectionGeneration && !manuallyPaused && isConnectionEnabled(platform)) { connect(platform, address); } } }, delaySeconds, TimeUnit.SECONDS); } private boolean isConnectionEnabled(Platform platform) { return platform == Platform.TWITCH ? config.streamerBotEnabled && (config.twitchEnabled || config.youtubeEnabled) : config.tikFinityEnabled && config.tiktokEnabled; } private boolean handleStreamerBotControl(JsonObject root) { String request = root.has("request") ? root.get("request").getAsString() : ""; if ("Hello".equalsIgnoreCase(request)) { JsonObject authentication = root.has("authentication") && root.get("authentication").isJsonObject() ? root.getAsJsonObject("authentication") : new JsonObject(); String salt = value(authentication, "salt"); String challenge = value(authentication, "challenge"); if (!salt.isBlank() && !challenge.isBlank() && config.streamerBotPassword.isBlank()) { statuses.put(Platform.TWITCH, "Error: Streamer.bot password required"); if (config.youtubeEnabled) statuses.put(Platform.YOUTUBE, "Error: Streamer.bot password required"); } else if (!config.streamerBotPassword.isBlank() && !salt.isBlank() && !challenge.isBlank()) { JsonObject authRequest = new JsonObject(); authRequest.addProperty("request", "Authenticate"); authenticationRequestId = UUID.randomUUID().toString(); authRequest.addProperty("id", authenticationRequestId); authRequest.addProperty("authentication", sha256(challenge + salt + config.streamerBotPassword)); WebSocket socket = sockets.get(Platform.TWITCH); if (socket != null) socket.sendText(authRequest.toString(), true); } else { WebSocket socket = sockets.get(Platform.TWITCH); if (socket != null) sendStreamerBotSubscription(socket); } return true; } if (root.has("id") && root.has("status")) { String status = value(root, "status"); if ("ok".equalsIgnoreCase(status)) { synchronized (this) { statuses.put(Platform.TWITCH, "Connected"); statuses.put(Platform.YOUTUBE, config.youtubeEnabled ? "Connected" : "Disabled"); } if (authenticationRequestId != null && authenticationRequestId.equals(value(root, "id"))) { authenticationRequestId = null; WebSocket socket = sockets.get(Platform.TWITCH); if (socket != null) sendStreamerBotSubscription(socket); } } else if (!status.isBlank()) { String reason = value(root, "error"); synchronized (this) { statuses.put(Platform.TWITCH, "Error: " + (reason.isBlank() ? status : reason)); if (config.youtubeEnabled) { statuses.put(Platform.YOUTUBE, "Error: " + (reason.isBlank() ? status : reason)); } } LOGGER.warning("Streamer.bot request returned status: " + root); } return true; } if (root.has("event")) return false; return true; } private String sha256(String value) { try { byte[] digest = MessageDigest.getInstance("SHA-256") .digest(value.getBytes(StandardCharsets.UTF_8)); StringBuilder result = new StringBuilder(digest.length * 2); for (byte item : digest) result.append(String.format(Locale.ROOT, "%02x", item)); return result.toString(); } catch (NoSuchAlgorithmException exception) { throw new IllegalStateException("SHA-256 is unavailable", exception); } } private String value(JsonObject object, String key) { return object.has(key) && !object.get(key).isJsonNull() ? object.get(key).getAsString() : ""; } private boolean isKnownEvent(StreamEvent event) { if (event.platform() == Platform.TIKTOK) { return switch (event.type()) { case "chat", "gift", "share", "follow", "like", "join", "subscribe" -> true; default -> false; }; } return event.platform() == Platform.TWITCH ? contains(TWITCH_EVENTS, event.type()) : contains(YOUTUBE_EVENTS, event.type()); } private boolean contains(String[] values, String target) { for (String value : values) if (value.equalsIgnoreCase(target)) return true; return false; } private void display(StreamEvent event, DisplayTarget target) { String text = config.format(event); Component component = ColorText.parse(text); switch (target) { case CHAT -> { if (client.gui != null) client.gui.getChat().addMessage(component); } case ACTION_BAR -> { if (client.player != null) client.player.displayClientMessage(component, true); } case TOAST -> SystemToast.addOrUpdate( client.getToasts(), SystemToast.SystemToastIds.TUTORIAL_HINT, Component.literal(event.platform() + " " + event.type()), component); case HIDDEN -> { } } } private final class Listener implements WebSocket.Listener { private final Platform platform; private final long generation; private final StringBuilder fragments = new StringBuilder(); private Listener(Platform platform, long generation) { this.platform = platform; this.generation = generation; } @Override public void onOpen(WebSocket webSocket) { synchronized (ConnectionManager.this) { if (generation != connectionGeneration) { webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "Configuration changed"); return; } sockets.put(platform, webSocket); boolean platformEnabled = platform == Platform.TIKTOK ? config.tiktokEnabled : config.twitchEnabled; statuses.put(platform, platformEnabled ? "Connected" : "Disabled"); retryCounts.remove(platform); scheduledRetries.remove(platform); if (platform == Platform.TWITCH) { statuses.put(Platform.YOUTUBE, config.youtubeEnabled ? "Connected" : "Disabled"); } } webSocket.request(1); } @Override public CompletionStage onText(WebSocket webSocket, CharSequence data, boolean last) { if (generation != connectionGeneration) return null; fragments.append(data); if (last) { receive(platform, fragments.toString()); fragments.setLength(0); } webSocket.request(1); return null; } @Override public CompletionStage onClose(WebSocket webSocket, int statusCode, String reason) { synchronized (ConnectionManager.this) { if (generation == connectionGeneration) { sockets.remove(platform); statuses.put(platform, "Disconnected"); if (platform == Platform.TWITCH && config.youtubeEnabled) { statuses.put(Platform.YOUTUBE, "Disconnected"); } scheduleReconnect(platform, platform == Platform.TWITCH ? config.streamerBotUrl : config.tikFinityUrl, generation); } } return null; } @Override public void onError(WebSocket webSocket, Throwable error) { synchronized (ConnectionManager.this) { if (generation == connectionGeneration) { statuses.put(platform, "Error: " + error.getMessage()); scheduleReconnect(platform, platform == Platform.TWITCH ? config.streamerBotUrl : config.tikFinityUrl, generation); } } LOGGER.log(Level.WARNING, platform + " WebSocket error", error); } } }